"""RunState class for serializing and resuming agent runs with human-in-the-loop support."""

from __future__ import annotations

import asyncio
import copy
import dataclasses
import json
import threading
from collections import deque
from collections.abc import Callable, Iterator, Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic, Literal, cast
from uuid import uuid4

from openai.types.responses import (
    ResponseComputerToolCall,
    ResponseCustomToolCall,
    ResponseFunctionToolCall,
    ResponseOutputMessage,
    ResponseOutputRefusal,
    ResponseOutputText,
    ResponseReasoningItem,
)
from openai.types.responses.response_input_param import (
    ComputerCallOutput,
    FunctionCallOutput,
    LocalShellCallOutput,
    McpApprovalResponse,
)
from openai.types.responses.response_output_item import (
    LocalShellCall,
    McpApprovalRequest,
    McpListTools,
)
from pydantic import TypeAdapter, ValidationError
from typing_extensions import TypeVar

from ._tool_identity import (
    FunctionToolLookupKey,
    NamedToolLookupKey,
    build_function_tool_lookup_map,
    deserialize_function_tool_lookup_key,
    get_function_tool_lookup_key,
    get_function_tool_lookup_key_for_tool,
    get_function_tool_namespace,
    get_function_tool_qualified_name,
    serialize_function_tool_lookup_key,
)
from .agent import Agent
from .exceptions import UserError
from .guardrail import (
    GuardrailFunctionOutput,
    InputGuardrail,
    InputGuardrailResult,
    OutputGuardrail,
    OutputGuardrailResult,
)
from .handoffs import Handoff
from .items import (
    CompactionItem,
    HandoffCallItem,
    HandoffOutputItem,
    MCPApprovalRequestItem,
    MCPApprovalResponseItem,
    MCPListToolsItem,
    MessageOutputItem,
    ModelResponse,
    ReasoningItem,
    RunItem,
    ToolApprovalItem,
    ToolCallItem,
    ToolCallOutputItem,
    ToolSearchCallItem,
    ToolSearchOutputItem,
    TResponseInputItem,
    coerce_tool_search_call_raw_item,
    coerce_tool_search_output_raw_item,
)
from .logger import logger
from .run_context import RunContextWrapper
from .sandbox.capabilities.capability import Capability
from .sandbox.session.base_sandbox_session import BaseSandboxSession
from .tool import (
    ApplyPatchTool,
    ComputerTool,
    CustomTool,
    FunctionTool,
    HostedMCPTool,
    LocalShellTool,
    ShellTool,
    ToolOrigin,
)
from .tool_guardrails import (
    AllowBehavior,
    RaiseExceptionBehavior,
    RejectContentBehavior,
    ToolGuardrailFunctionOutput,
    ToolInputGuardrail,
    ToolInputGuardrailResult,
    ToolOutputGuardrail,
    ToolOutputGuardrailResult,
)
from .tracing.traces import Trace, TraceState
from .usage import deserialize_usage, serialize_usage
from .util._json import _to_dump_compatible

if TYPE_CHECKING:
    from .guardrail import InputGuardrailResult, OutputGuardrailResult
    from .items import ModelResponse, RunItem
    from .run_internal.run_steps import (
        NextStepInterruption,
        ProcessedResponse,
    )

TContext = TypeVar("TContext", default=Any)
TAgent = TypeVar("TAgent", bound="Agent[Any]", default="Agent[Any]")
ContextOverride = Mapping[str, Any] | RunContextWrapper[Any]
ContextSerializer = Callable[[Any], Mapping[str, Any]]
ContextDeserializer = Callable[[Mapping[str, Any]], Any]


# RunState schema policy.
# 1. Keep schema versions shipped in releases readable.
# 2. Unreleased schema versions may be renumbered or squashed before release when their
#    intermediate snapshots are intentionally unsupported.
# 3. to_json() always emits CURRENT_SCHEMA_VERSION.
# 4. Forward compatibility is intentionally fail-fast (older SDKs reject newer or unsupported
#    versions).
CURRENT_SCHEMA_VERSION = "1.9"
# Keep this mapping in chronological order. Every schema bump must add a one-line summary here.
SCHEMA_VERSION_SUMMARIES: dict[str, str] = {
    "1.0": "Initial RunState snapshot format for HITL pause/resume flows.",
    "1.1": "Same payload as 1.0, but introduces explicit backward-read support policy.",
    "1.2": "Persists reasoning_item_id_policy for resumed and streamed follow-up turns.",
    "1.3": "Updates resumed trace semantics to reattach traces without duplicate starts.",
    "1.4": "Stores request_id alongside each serialized model response.",
    "1.5": "Renumbered unreleased baseline for tool-search snapshots and richer tool metadata.",
    "1.6": "Persists explicit approval rejection messages across resume flows.",
    "1.7": (
        "Persists duplicate-name agent identities across agent-owned state "
        "and sandbox resume state."
    ),
    "1.8": "Persists SDK-generated prompt cache keys across resume flows.",
    "1.9": "Persists pending custom tool calls and tool origin metadata across resume flows.",
}
SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES)

if CURRENT_SCHEMA_VERSION not in SCHEMA_VERSION_SUMMARIES:
    raise AssertionError(
        "CURRENT_SCHEMA_VERSION must have a matching entry in SCHEMA_VERSION_SUMMARIES."
    )

_missing_schema_version_summaries = [
    version for version, summary in SCHEMA_VERSION_SUMMARIES.items() if not summary.strip()
]
if _missing_schema_version_summaries:
    raise AssertionError(
        "Every supported RunState schema version must have a non-empty summary. "
        f"Missing summaries: {', '.join(_missing_schema_version_summaries)}"
    )

_FUNCTION_OUTPUT_ADAPTER: TypeAdapter[FunctionCallOutput] = TypeAdapter(FunctionCallOutput)
_COMPUTER_OUTPUT_ADAPTER: TypeAdapter[ComputerCallOutput] = TypeAdapter(ComputerCallOutput)
_LOCAL_SHELL_OUTPUT_ADAPTER: TypeAdapter[LocalShellCallOutput] = TypeAdapter(LocalShellCallOutput)
_TOOL_CALL_OUTPUT_UNION_ADAPTER: TypeAdapter[
    FunctionCallOutput | ComputerCallOutput | LocalShellCallOutput
] = TypeAdapter(FunctionCallOutput | ComputerCallOutput | LocalShellCallOutput)
_MCP_APPROVAL_RESPONSE_ADAPTER: TypeAdapter[McpApprovalResponse] = TypeAdapter(McpApprovalResponse)
_HANDOFF_OUTPUT_ADAPTER: TypeAdapter[TResponseInputItem] = TypeAdapter(TResponseInputItem)
_LOCAL_SHELL_CALL_ADAPTER: TypeAdapter[LocalShellCall] = TypeAdapter(LocalShellCall)
_MISSING_CONTEXT_SENTINEL = object()
_ALLOWED_MISSING_MESSAGE_FIELDS = frozenset({"status"})


def _deserialize_tool_origin(data: Any) -> ToolOrigin | None:
    """Best-effort deserialization for optional tool origin metadata."""
    return ToolOrigin.from_json_dict(data)


@dataclass
class RunState(Generic[TContext, TAgent]):
    """Serializable snapshot of an agent run, including context, usage, and interruptions.

    ``RunState`` is the durable pause/resume boundary for human-in-the-loop flows. It stores
    enough information to continue an interrupted run, including model responses, generated
    items, approval state, and optional server-managed conversation identifiers.

    Context serialization is intentionally conservative:

    - Mapping contexts round-trip directly.
    - Custom contexts may require a serializer and deserializer.
    - When no safe serializer is available, the snapshot is still written but emits warnings and
      records metadata describing what is required to rebuild the original context type.
    """

    _current_turn: int = 0
    """Current turn number in the conversation."""

    _current_agent: TAgent | None = None
    """The agent currently handling the conversation."""

    _starting_agent: TAgent | None = field(default=None, repr=False)
    """The root agent used to derive stable duplicate-name identities during resume."""

    _original_input: str | list[Any] = field(default_factory=list)
    """Original user input prior to any processing."""

    _model_responses: list[ModelResponse] = field(default_factory=list)
    """Responses from the model so far."""

    _context: RunContextWrapper[TContext] | None = None
    """Run context tracking approvals, usage, and other metadata."""

    _generated_items: list[RunItem] = field(default_factory=list)
    """Items used to build model input when resuming; may be filtered by handoffs."""

    _session_items: list[RunItem] = field(default_factory=list)
    """Full, unfiltered run items for session history."""

    _max_turns: int = 10
    """Maximum allowed turns before forcing termination."""

    _conversation_id: str | None = None
    """Conversation identifier for server-managed conversation tracking."""

    _previous_response_id: str | None = None
    """Response identifier of the last server-managed response."""

    _auto_previous_response_id: bool = False
    """Whether the previous response id should be automatically tracked."""

    _generated_prompt_cache_key: str | None = None
    """SDK-generated prompt cache key to preserve across resume flows."""

    _reasoning_item_id_policy: Literal["preserve", "omit"] | None = None
    """How reasoning item IDs are represented in next-turn model input."""

    _input_guardrail_results: list[InputGuardrailResult] = field(default_factory=list)
    """Results from input guardrails applied to the run."""

    _output_guardrail_results: list[OutputGuardrailResult] = field(default_factory=list)
    """Results from output guardrails applied to the run."""

    _tool_input_guardrail_results: list[ToolInputGuardrailResult] = field(default_factory=list)
    """Results from tool input guardrails applied during the run."""

    _tool_output_guardrail_results: list[ToolOutputGuardrailResult] = field(default_factory=list)
    """Results from tool output guardrails applied during the run."""

    _current_step: NextStepInterruption | None = None
    """Current step if the run is interrupted (e.g., for tool approval)."""

    _last_processed_response: ProcessedResponse | None = None
    """The last processed model response. This is needed for resuming from interruptions."""

    _generated_items_last_processed_marker: str | None = field(default=None, repr=False)
    """Tracks whether _generated_items already include the current last_processed_response."""

    _current_turn_persisted_item_count: int = 0
    """Tracks how many items from this turn were already written to the session."""

    _tool_use_tracker_snapshot: dict[str, list[str]] = field(default_factory=dict)
    """Serialized snapshot of the AgentToolUseTracker (agent name -> tools used)."""

    _trace_state: TraceState | None = field(default=None, repr=False)
    """Serialized trace metadata for resuming tracing context."""

    _agent_tool_state_scope_id: str | None = field(default=None, repr=False)
    """Private scope id used to isolate agent-tool pending state per RunState instance."""

    _sandbox: dict[str, Any] | None = field(default=None, repr=False)
    """Serialized sandbox resume payload for sandbox-aware runs."""

    _schema_version: str = field(default=CURRENT_SCHEMA_VERSION, repr=False)
    """Schema version the snapshot was loaded from for schema-gated resume compatibility."""

    def __init__(
        self,
        context: RunContextWrapper[TContext],
        original_input: str | list[Any],
        starting_agent: TAgent,
        max_turns: int = 10,
        *,
        conversation_id: str | None = None,
        previous_response_id: str | None = None,
        auto_previous_response_id: bool = False,
    ):
        """Initialize a new RunState."""
        self._context = context
        self._original_input = _clone_original_input(original_input)
        self._starting_agent = starting_agent
        self._current_agent = starting_agent
        self._max_turns = max_turns
        self._conversation_id = conversation_id
        self._previous_response_id = previous_response_id
        self._auto_previous_response_id = auto_previous_response_id
        self._generated_prompt_cache_key = None
        self._reasoning_item_id_policy = None
        self._model_responses = []
        self._generated_items = []
        self._session_items = []
        self._input_guardrail_results = []
        self._output_guardrail_results = []
        self._tool_input_guardrail_results = []
        self._tool_output_guardrail_results = []
        self._current_step = None
        self._current_turn = 0
        self._last_processed_response = None
        self._generated_items_last_processed_marker = None
        self._current_turn_persisted_item_count = 0
        self._tool_use_tracker_snapshot = {}
        self._trace_state = None
        self._sandbox = None
        self._schema_version = CURRENT_SCHEMA_VERSION
        from .agent_tool_state import get_agent_tool_state_scope

        self._agent_tool_state_scope_id = get_agent_tool_state_scope(context)

    def get_interruptions(self) -> list[ToolApprovalItem]:
        """Return pending interruptions if the current step is an interruption."""
        # Import at runtime to avoid circular import
        from .run_internal.run_steps import NextStepInterruption

        if self._current_step is None or not isinstance(self._current_step, NextStepInterruption):
            return []
        return self._current_step.interruptions

    def approve(self, approval_item: ToolApprovalItem, always_approve: bool = False) -> None:
        """Approve a tool call and rerun with this state to continue."""
        if self._context is None:
            raise UserError("Cannot approve tool: RunState has no context")
        self._context.approve_tool(approval_item, always_approve=always_approve)

    def reject(
        self,
        approval_item: ToolApprovalItem,
        always_reject: bool = False,
        *,
        rejection_message: str | None = None,
    ) -> None:
        """Reject a tool call and rerun with this state to continue.

        When ``rejection_message`` is provided, that exact text is sent back to the model when the
        run resumes. Otherwise the run-level tool error formatter or the SDK default message is
        used.
        """
        if self._context is None:
            raise UserError("Cannot reject tool: RunState has no context")
        self._context.reject_tool(
            approval_item,
            always_reject=always_reject,
            rejection_message=rejection_message,
        )

    def _serialize_approvals(self) -> dict[str, dict[str, Any]]:
        """Serialize approval records into a JSON-friendly mapping."""
        if self._context is None:
            return {}
        approvals_dict: dict[str, dict[str, Any]] = {}
        for tool_name, record in self._context._approvals.items():
            approvals_dict[tool_name] = {
                "approved": record.approved
                if isinstance(record.approved, bool)
                else list(record.approved),
                "rejected": record.rejected
                if isinstance(record.rejected, bool)
                else list(record.rejected),
            }
            if record.rejection_messages:
                approvals_dict[tool_name]["rejection_messages"] = dict(record.rejection_messages)
            if record.sticky_rejection_message is not None:
                approvals_dict[tool_name]["sticky_rejection_message"] = (
                    record.sticky_rejection_message
                )
        return approvals_dict

    def _serialize_model_responses(self) -> list[dict[str, Any]]:
        """Serialize model responses."""
        return [
            {
                "usage": serialize_usage(resp.usage),
                "output": [_serialize_raw_item_value(item) for item in resp.output],
                "response_id": resp.response_id,
                "request_id": resp.request_id,
            }
            for resp in self._model_responses
        ]

    def _serialize_original_input(self) -> str | list[Any]:
        """Normalize original input into the shape expected by Responses API."""
        if not isinstance(self._original_input, list):
            return self._original_input

        normalized_items = []
        for item in self._original_input:
            normalized_item = _serialize_raw_item_value(item)
            if isinstance(normalized_item, dict):
                normalized_item = dict(normalized_item)
                role = normalized_item.get("role")
                if role == "assistant":
                    content = normalized_item.get("content")
                    if isinstance(content, str):
                        normalized_item["content"] = [{"type": "output_text", "text": content}]
                    if "status" not in normalized_item:
                        normalized_item["status"] = "completed"
            normalized_items.append(normalized_item)
        return normalized_items

    def _serialize_context_payload(
        self,
        *,
        context_serializer: ContextSerializer | None = None,
        strict_context: bool = False,
    ) -> tuple[dict[str, Any] | None, dict[str, Any]]:
        """Validate and serialize the stored run context.

        The returned metadata captures how the context was serialized so restore-time code can
        decide whether a deserializer or override is required. This lets RunState remain durable
        for simple mapping contexts without silently pretending that richer custom objects can be
        reconstructed automatically.
        """
        if self._context is None:
            return None, _build_context_meta(
                None,
                serialized_via="none",
                requires_deserializer=False,
                omitted=False,
            )

        raw_context_payload = self._context.context
        if raw_context_payload is None:
            return None, _build_context_meta(
                raw_context_payload,
                serialized_via="none",
                requires_deserializer=False,
                omitted=False,
            )

        if isinstance(raw_context_payload, Mapping):
            return (
                dict(raw_context_payload),
                _build_context_meta(
                    raw_context_payload,
                    serialized_via="mapping",
                    requires_deserializer=False,
                    omitted=False,
                ),
            )

        if strict_context and context_serializer is None:
            # Avoid silently dropping non-mapping context data when strict mode is requested.
            raise UserError(
                "RunState serialization requires context to be a mapping when strict_context "
                "is True. Provide context_serializer to serialize custom contexts."
            )

        if context_serializer is not None:
            try:
                serialized = context_serializer(raw_context_payload)
            except Exception as exc:
                raise UserError(
                    "Context serializer failed while serializing RunState context."
                ) from exc
            if not isinstance(serialized, Mapping):
                raise UserError("Context serializer must return a mapping.")
            return (
                dict(serialized),
                _build_context_meta(
                    raw_context_payload,
                    serialized_via="context_serializer",
                    requires_deserializer=True,
                    omitted=False,
                ),
            )

        if hasattr(raw_context_payload, "model_dump"):
            try:
                serialized = raw_context_payload.model_dump(exclude_unset=True)
            except TypeError:
                serialized = raw_context_payload.model_dump()
            if not isinstance(serialized, Mapping):
                raise UserError("RunState context model_dump must return a mapping.")
            # We can persist the data, but the original type is lost unless the caller rebuilds it.
            logger.warning(
                "RunState context was serialized from a Pydantic model. "
                "Provide context_deserializer or context_override to restore the original type."
            )
            return (
                dict(serialized),
                _build_context_meta(
                    raw_context_payload,
                    serialized_via="model_dump",
                    requires_deserializer=True,
                    omitted=False,
                ),
            )

        if dataclasses.is_dataclass(raw_context_payload):
            serialized = dataclasses.asdict(cast(Any, raw_context_payload))
            if not isinstance(serialized, Mapping):
                raise UserError("RunState dataclass context must serialize to a mapping.")
            # Dataclass instances serialize to dicts, so reconstruction requires a deserializer.
            logger.warning(
                "RunState context was serialized from a dataclass. "
                "Provide context_deserializer or context_override to restore the original type."
            )
            return (
                dict(serialized),
                _build_context_meta(
                    raw_context_payload,
                    serialized_via="asdict",
                    requires_deserializer=True,
                    omitted=False,
                ),
            )

        # Fall back to an empty dict so the run state remains serializable, but
        # explicitly warn because the original context will be unavailable on restore.
        logger.warning(
            "RunState context of type %s is not serializable; storing empty context. "
            "Provide context_serializer to preserve it.",
            type(raw_context_payload).__name__,
        )
        return (
            {},
            _build_context_meta(
                raw_context_payload,
                serialized_via="omitted",
                requires_deserializer=True,
                omitted=True,
            ),
        )

    def _serialize_tool_input(self, tool_input: Any) -> Any:
        """Normalize tool input for JSON serialization."""
        if tool_input is None:
            return None

        if dataclasses.is_dataclass(tool_input):
            return dataclasses.asdict(cast(Any, tool_input))

        if hasattr(tool_input, "model_dump"):
            try:
                serialized = tool_input.model_dump(exclude_unset=True)
            except TypeError:
                serialized = tool_input.model_dump()
            return _to_dump_compatible(serialized)

        return _to_dump_compatible(tool_input)

    def _current_generated_items_merge_marker(self) -> str | None:
        """Return a marker for the processed response already reflected in _generated_items."""
        if not (self._last_processed_response and self._last_processed_response.new_items):
            return None

        latest_response_id = (
            self._model_responses[-1].response_id if self._model_responses else None
        )
        agent_identity_keys_by_id = (
            _build_agent_identity_keys_by_id(cast(Agent[Any], self._starting_agent))
            if self._starting_agent is not None
            else None
        )
        serialized_items = [
            self._serialize_item(item, agent_identity_keys_by_id=agent_identity_keys_by_id)
            for item in self._last_processed_response.new_items
        ]
        return json.dumps(
            {
                "current_turn": self._current_turn,
                "last_response_id": latest_response_id,
                "new_items": serialized_items,
            },
            sort_keys=True,
            default=str,
        )

    def _mark_generated_items_merged_with_last_processed(self) -> None:
        """Remember that _generated_items already include the current processed response."""
        self._generated_items_last_processed_marker = self._current_generated_items_merge_marker()

    def _clear_generated_items_last_processed_marker(self) -> None:
        """Forget any prior merge marker after _generated_items is replaced."""
        self._generated_items_last_processed_marker = None

    def _merge_generated_items_with_processed(self) -> list[RunItem]:
        """Merge persisted and newly processed items without duplication."""
        generated_items = list(self._generated_items)
        if not (self._last_processed_response and self._last_processed_response.new_items):
            return generated_items

        current_merge_marker = self._current_generated_items_merge_marker()
        if (
            current_merge_marker is not None
            and self._generated_items_last_processed_marker == current_merge_marker
        ):
            return generated_items

        seen_id_types: set[tuple[str, str]] = set()
        seen_call_ids: set[str] = set()
        seen_call_id_types: set[tuple[str, str]] = set()

        def _id_type_call(item: Any) -> tuple[str | None, str | None, str | None]:
            item_id = None
            item_type = None
            call_id = None
            if hasattr(item, "raw_item"):
                raw = item.raw_item
                if isinstance(raw, dict):
                    item_id = raw.get("id")
                    item_type = raw.get("type")
                    call_id = raw.get("call_id")
                else:
                    item_id = _get_attr(raw, "id")
                    item_type = _get_attr(raw, "type")
                    call_id = _get_attr(raw, "call_id")
            if item_id is None and hasattr(item, "id"):
                item_id = _get_attr(item, "id")
            if item_type is None and hasattr(item, "type"):
                item_type = _get_attr(item, "type")
            return item_id, item_type, call_id

        for existing in generated_items:
            item_id, item_type, call_id = _id_type_call(existing)
            if item_id and item_type:
                seen_id_types.add((item_id, item_type))
            if call_id and item_type:
                seen_call_id_types.add((call_id, item_type))
            elif call_id:
                seen_call_ids.add(call_id)

        for new_item in self._last_processed_response.new_items:
            item_id, item_type, call_id = _id_type_call(new_item)
            if call_id and item_type:
                if (call_id, item_type) in seen_call_id_types:
                    continue
            elif call_id and call_id in seen_call_ids:
                continue
            if item_id and item_type and (item_id, item_type) in seen_id_types:
                continue
            if item_id and item_type:
                seen_id_types.add((item_id, item_type))
            if call_id and item_type:
                seen_call_id_types.add((call_id, item_type))
            elif call_id:
                seen_call_ids.add(call_id)
            generated_items.append(new_item)

        if current_merge_marker is not None:
            self._generated_items_last_processed_marker = current_merge_marker
        return generated_items

    def to_json(
        self,
        *,
        context_serializer: ContextSerializer | None = None,
        strict_context: bool = False,
        include_tracing_api_key: bool = False,
    ) -> dict[str, Any]:
        """Serializes the run state to a JSON-compatible dictionary.

        This method is used to serialize the run state to a dictionary that can be used to
        resume the run later.

        Args:
            context_serializer: Optional function to serialize non-mapping context values.
            strict_context: When True, require mapping contexts or a context_serializer.
            include_tracing_api_key: When True, include the tracing API key in the trace payload.

        Returns:
            A dictionary representation of the run state.

        Raises:
            UserError: If required state (agent, context) is missing.
        """
        if self._current_agent is None:
            raise UserError("Cannot serialize RunState: No current agent")
        if self._context is None:
            raise UserError("Cannot serialize RunState: No context")

        approvals_dict = self._serialize_approvals()
        model_responses = self._serialize_model_responses()
        original_input_serialized = self._serialize_original_input()
        context_payload, context_meta = self._serialize_context_payload(
            context_serializer=context_serializer,
            strict_context=strict_context,
        )

        context_entry: dict[str, Any] = {
            "usage": serialize_usage(self._context.usage),
            "approvals": approvals_dict,
            "context": context_payload,
            # Preserve metadata so deserialization can warn when context types were erased.
            "context_meta": context_meta,
        }
        tool_input = self._serialize_tool_input(self._context.tool_input)
        if tool_input is not None:
            context_entry["tool_input"] = tool_input

        agent_identity_keys_by_id = (
            _build_agent_identity_keys_by_id(cast(Agent[Any], self._starting_agent))
            if self._starting_agent is not None
            else None
        )
        current_agent_entry = _serialize_agent_reference(
            cast(Agent[Any], self._current_agent),
            agent_identity_keys_by_id=agent_identity_keys_by_id,
        )

        result = {
            "$schemaVersion": CURRENT_SCHEMA_VERSION,
            "current_turn": self._current_turn,
            "current_agent": current_agent_entry,
            "original_input": original_input_serialized,
            "model_responses": model_responses,
            "context": context_entry,
            "tool_use_tracker": copy.deepcopy(self._tool_use_tracker_snapshot),
            "max_turns": self._max_turns,
            "no_active_agent_run": True,
            "input_guardrail_results": _serialize_guardrail_results(
                self._input_guardrail_results,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            ),
            "output_guardrail_results": _serialize_guardrail_results(
                self._output_guardrail_results,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            ),
            "tool_input_guardrail_results": _serialize_tool_guardrail_results(
                self._tool_input_guardrail_results, type_label="tool_input"
            ),
            "tool_output_guardrail_results": _serialize_tool_guardrail_results(
                self._tool_output_guardrail_results, type_label="tool_output"
            ),
            "conversation_id": self._conversation_id,
            "previous_response_id": self._previous_response_id,
            "auto_previous_response_id": self._auto_previous_response_id,
            "generated_prompt_cache_key": self._generated_prompt_cache_key,
            "reasoning_item_id_policy": self._reasoning_item_id_policy,
        }

        generated_items = self._merge_generated_items_with_processed()
        result["generated_items"] = [
            self._serialize_item(item, agent_identity_keys_by_id=agent_identity_keys_by_id)
            for item in generated_items
        ]
        result["session_items"] = [
            self._serialize_item(item, agent_identity_keys_by_id=agent_identity_keys_by_id)
            for item in list(self._session_items)
        ]
        result["current_step"] = self._serialize_current_step()
        result["last_model_response"] = _serialize_last_model_response(model_responses)
        result["last_processed_response"] = (
            self._serialize_processed_response(
                self._last_processed_response,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
                context_serializer=context_serializer,
                strict_context=strict_context,
                include_tracing_api_key=include_tracing_api_key,
            )
            if self._last_processed_response
            else None
        )
        result["current_turn_persisted_item_count"] = self._current_turn_persisted_item_count
        result["trace"] = self._serialize_trace_data(
            include_tracing_api_key=include_tracing_api_key
        )
        if self._sandbox is not None:
            result["sandbox"] = copy.deepcopy(self._sandbox)

        return result

    def _serialize_processed_response(
        self,
        processed_response: ProcessedResponse,
        *,
        agent_identity_keys_by_id: Mapping[int, str] | None = None,
        context_serializer: ContextSerializer | None = None,
        strict_context: bool = False,
        include_tracing_api_key: bool = False,
    ) -> dict[str, Any]:
        """Serialize a ProcessedResponse to JSON format.

        Args:
            processed_response: The ProcessedResponse to serialize.

        Returns:
            A dictionary representation of the ProcessedResponse.
        """

        action_groups = _serialize_tool_action_groups(processed_response)
        _serialize_pending_nested_agent_tool_runs(
            parent_state=self,
            function_entries=action_groups.get("functions", []),
            function_runs=processed_response.functions,
            scope_id=self._agent_tool_state_scope_id,
            context_serializer=context_serializer,
            strict_context=strict_context,
            include_tracing_api_key=include_tracing_api_key,
        )

        interruptions_data = [
            _serialize_tool_approval_interruption(
                interruption,
                include_tool_name=True,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            )
            for interruption in processed_response.interruptions
            if isinstance(interruption, ToolApprovalItem)
        ]

        return {
            "new_items": [
                self._serialize_item(item, agent_identity_keys_by_id=agent_identity_keys_by_id)
                for item in processed_response.new_items
            ],
            "tools_used": processed_response.tools_used,
            **action_groups,
            "interruptions": interruptions_data,
        }

    def _serialize_current_step(self) -> dict[str, Any] | None:
        """Serialize the current step if it's an interruption."""
        # Import at runtime to avoid circular import
        from .run_internal.run_steps import NextStepInterruption

        agent_identity_keys_by_id = (
            _build_agent_identity_keys_by_id(cast(Agent[Any], self._starting_agent))
            if self._starting_agent is not None
            else None
        )

        if self._current_step is None or not isinstance(self._current_step, NextStepInterruption):
            return None

        interruptions_data = [
            _serialize_tool_approval_interruption(
                item,
                include_tool_name=item.tool_name is not None,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            )
            for item in self._current_step.interruptions
            if isinstance(item, ToolApprovalItem)
        ]

        return {
            "type": "next_step_interruption",
            "data": {
                "interruptions": interruptions_data,
            },
        }

    def _serialize_item(
        self,
        item: RunItem,
        *,
        agent_identity_keys_by_id: Mapping[int, str] | None = None,
    ) -> dict[str, Any]:
        """Serialize a run item to JSON-compatible dict."""
        raw_item_dict: Any = _serialize_raw_item_value(item.raw_item)

        result: dict[str, Any] = {
            "type": item.type,
            "raw_item": raw_item_dict,
            "agent": _serialize_agent_reference(
                item.agent,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            ),
        }

        # Add additional fields based on item type
        if hasattr(item, "output"):
            serialized_output = item.output
            try:
                if hasattr(serialized_output, "model_dump"):
                    serialized_output = serialized_output.model_dump(exclude_unset=True)
                elif dataclasses.is_dataclass(serialized_output):
                    serialized_output = dataclasses.asdict(serialized_output)  # type: ignore[arg-type]
                serialized_output = _ensure_json_compatible(serialized_output)
            except Exception:
                serialized_output = str(item.output)
            result["output"] = serialized_output
        if hasattr(item, "source_agent"):
            result["source_agent"] = _serialize_agent_reference(
                item.source_agent,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            )
        if hasattr(item, "target_agent"):
            result["target_agent"] = _serialize_agent_reference(
                item.target_agent,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            )
        if hasattr(item, "tool_name") and item.tool_name is not None:
            result["tool_name"] = item.tool_name
        if hasattr(item, "tool_namespace") and item.tool_namespace is not None:
            result["tool_namespace"] = item.tool_namespace
        tool_lookup_key = serialize_function_tool_lookup_key(getattr(item, "tool_lookup_key", None))
        if tool_lookup_key is not None:
            result["tool_lookup_key"] = tool_lookup_key
        if getattr(item, "_allow_bare_name_alias", False):
            result["allow_bare_name_alias"] = True
        if hasattr(item, "description") and item.description is not None:
            result["description"] = item.description
        if hasattr(item, "title") and item.title is not None:
            result["title"] = item.title
        tool_origin = getattr(item, "tool_origin", None)
        if isinstance(tool_origin, ToolOrigin):
            result["tool_origin"] = tool_origin.to_json_dict()

        return result

    def _lookup_function_name(self, call_id: str) -> str:
        """Attempt to find the function name for the provided call_id."""
        if not call_id:
            return ""

        def _extract_name(raw: Any) -> str | None:
            if isinstance(raw, dict):
                candidate_call_id = cast(str | None, raw.get("call_id"))
                if candidate_call_id == call_id:
                    name_value = raw.get("name", "")
                    return str(name_value) if name_value else ""
            else:
                candidate_call_id = cast(str | None, _get_attr(raw, "call_id"))
                if candidate_call_id == call_id:
                    name_value = _get_attr(raw, "name", "")
                    return str(name_value) if name_value else ""
            return None

        # Search generated items first
        for run_item in self._generated_items:
            if run_item.type != "tool_call_item":
                continue
            name = _extract_name(run_item.raw_item)
            if name is not None:
                return name

        # Inspect last processed response
        if self._last_processed_response is not None:
            for run_item in self._last_processed_response.new_items:
                if run_item.type != "tool_call_item":
                    continue
                name = _extract_name(run_item.raw_item)
                if name is not None:
                    return name

        # Finally, inspect the original input list where the function call originated
        if isinstance(self._original_input, list):
            for input_item in self._original_input:
                if not isinstance(input_item, dict):
                    continue
                if input_item.get("type") != "function_call":
                    continue
                item_call_id = cast(str | None, input_item.get("call_id"))
                if item_call_id == call_id:
                    name_value = input_item.get("name", "")
                    return str(name_value) if name_value else ""

        return ""

    def to_string(
        self,
        *,
        context_serializer: ContextSerializer | None = None,
        strict_context: bool = False,
        include_tracing_api_key: bool = False,
    ) -> str:
        """Serializes the run state to a JSON string.

        Args:
            include_tracing_api_key: When True, include the tracing API key in the trace payload.

        Returns:
            JSON string representation of the run state.
        """
        return json.dumps(
            self.to_json(
                context_serializer=context_serializer,
                strict_context=strict_context,
                include_tracing_api_key=include_tracing_api_key,
            ),
            indent=2,
        )

    def set_trace(self, trace: Trace | None) -> None:
        """Capture trace metadata for serialization/resumption."""
        self._trace_state = TraceState.from_trace(trace)

    def _serialize_trace_data(self, *, include_tracing_api_key: bool) -> dict[str, Any] | None:
        if not self._trace_state:
            return None
        return self._trace_state.to_json(include_tracing_api_key=include_tracing_api_key)

    def set_tool_use_tracker_snapshot(self, snapshot: Mapping[str, Sequence[str]] | None) -> None:
        """Store a copy of the serialized tool-use tracker data."""
        if not snapshot:
            self._tool_use_tracker_snapshot = {}
            return

        normalized: dict[str, list[str]] = {}
        for agent_name, tools in snapshot.items():
            if not isinstance(agent_name, str):
                continue
            normalized[agent_name] = [tool for tool in tools if isinstance(tool, str)]
        self._tool_use_tracker_snapshot = normalized

    def set_reasoning_item_id_policy(self, policy: Literal["preserve", "omit"] | None) -> None:
        """Store how reasoning item IDs should appear in next-turn model input."""
        self._reasoning_item_id_policy = policy

    def get_tool_use_tracker_snapshot(self) -> dict[str, list[str]]:
        """Return a defensive copy of the tool-use tracker snapshot."""
        return {
            agent_name: list(tool_names)
            for agent_name, tool_names in self._tool_use_tracker_snapshot.items()
        }

    @staticmethod
    async def from_string(
        initial_agent: Agent[Any],
        state_string: str,
        *,
        context_override: ContextOverride | None = None,
        context_deserializer: ContextDeserializer | None = None,
        strict_context: bool = False,
    ) -> RunState[Any, Agent[Any]]:
        """Deserializes a run state from a JSON string.

        This method is used to deserialize a run state from a string that was serialized using
        the `to_string()` method.

        Args:
            initial_agent: The initial agent (used to build agent map for resolution).
            state_string: The JSON string to deserialize.
            context_override: Optional context mapping or RunContextWrapper to use instead of the
                serialized context.
            context_deserializer: Optional function to rebuild non-mapping context values.
            strict_context: When True, require a deserializer or override for non-mapping contexts.

        Returns:
            A reconstructed RunState instance.

        Raises:
            UserError: If the string is invalid JSON or has incompatible schema version.
        """
        try:
            state_json = json.loads(state_string)
        except json.JSONDecodeError as e:
            raise UserError(f"Failed to parse run state JSON: {e}") from e

        return await RunState.from_json(
            initial_agent=initial_agent,
            state_json=state_json,
            context_override=context_override,
            context_deserializer=context_deserializer,
            strict_context=strict_context,
        )

    @staticmethod
    async def from_json(
        initial_agent: Agent[Any],
        state_json: dict[str, Any],
        *,
        context_override: ContextOverride | None = None,
        context_deserializer: ContextDeserializer | None = None,
        strict_context: bool = False,
    ) -> RunState[Any, Agent[Any]]:
        """Deserializes a run state from a JSON dictionary.

        This method is used to deserialize a run state from a dict that was created using
        the `to_json()` method.

        Args:
            initial_agent: The initial agent (used to build agent map for resolution).
            state_json: The JSON dictionary to deserialize.
            context_override: Optional context mapping or RunContextWrapper to use instead of the
                serialized context.
            context_deserializer: Optional function to rebuild non-mapping context values.
            strict_context: When True, require a deserializer or override for non-mapping contexts.

        Returns:
            A reconstructed RunState instance.

        Raises:
            UserError: If the dict has incompatible schema version.
        """
        return await _build_run_state_from_json(
            initial_agent=initial_agent,
            state_json=state_json,
            context_override=context_override,
            context_deserializer=context_deserializer,
            strict_context=strict_context,
        )


# --------------------------
# Private helpers
# --------------------------


def _get_attr(obj: Any, attr: str, default: Any = None) -> Any:
    """Return attribute value if present, otherwise the provided default."""
    return getattr(obj, attr, default)


def _describe_context_type(value: Any) -> str:
    """Summarize a context object for serialization metadata."""
    if value is None:
        return "none"
    if isinstance(value, Mapping):
        return "mapping"
    if hasattr(value, "model_dump"):
        return "pydantic"
    if dataclasses.is_dataclass(value):
        return "dataclass"
    return "custom"


def _context_class_path(value: Any) -> str | None:
    """Return module and qualname for debugging purposes."""
    if value is None:
        return None
    cls = value.__class__
    module = getattr(cls, "__module__", "")
    qualname = getattr(cls, "__qualname__", "")
    if not module or not qualname:
        return None
    return f"{module}:{qualname}"


def _build_context_meta(
    original_context: Any,
    *,
    serialized_via: str,
    requires_deserializer: bool,
    omitted: bool,
) -> dict[str, Any]:
    """Capture context serialization metadata for debugging and recovery hints."""
    original_type = _describe_context_type(original_context)
    meta: dict[str, Any] = {
        "original_type": original_type,
        "serialized_via": serialized_via,
        "requires_deserializer": requires_deserializer,
        "omitted": omitted,
    }
    class_path = _context_class_path(original_context)
    if class_path and original_type not in {"mapping", "none"}:
        # Store the class path for reference only; never auto-import it for safety.
        meta["class_path"] = class_path
    return meta


def _context_meta_requires_deserializer(context_meta: Mapping[str, Any] | None) -> bool:
    """Return True when metadata indicates a non-mapping context needs help to restore."""
    if not isinstance(context_meta, Mapping):
        return False
    if context_meta.get("omitted"):
        return True
    return bool(context_meta.get("requires_deserializer"))


def _context_meta_warning_message(context_meta: Mapping[str, Any] | None) -> str:
    """Build a warning message describing context deserialization requirements."""
    if not isinstance(context_meta, Mapping):
        return (
            "RunState context was serialized from a custom type; provide context_deserializer "
            "or context_override to restore it."
        )
    original_type = context_meta.get("original_type") or "custom"
    class_path = context_meta.get("class_path")
    type_label = f"{original_type} ({class_path})" if class_path else str(original_type)
    if context_meta.get("omitted"):
        return (
            "RunState context was omitted during serialization for "
            f"{type_label}; provide context_override to supply it."
        )
    return (
        "RunState context was serialized from "
        f"{type_label}; provide context_deserializer or context_override to restore it."
    )


def _transform_field_names(
    data: dict[str, Any] | list[Any] | Any, field_map: Mapping[str, str]
) -> Any:
    """Recursively remap field names using the provided mapping."""
    if isinstance(data, dict):
        transformed: dict[str, Any] = {}
        for key, value in data.items():
            mapped_key = field_map.get(key, key)
            if isinstance(value, dict | list):
                transformed[mapped_key] = _transform_field_names(value, field_map)
            else:
                transformed[mapped_key] = value
        return transformed

    if isinstance(data, list):
        return [
            _transform_field_names(item, field_map) if isinstance(item, dict | list) else item
            for item in data
        ]

    return data


def _serialize_raw_item_value(raw_item: Any) -> Any:
    """Return a serializable representation of a raw item."""
    if hasattr(raw_item, "model_dump"):
        return raw_item.model_dump(exclude_unset=True)
    if isinstance(raw_item, dict):
        return dict(raw_item)
    return raw_item


def _serialize_agent_reference(
    agent: Agent[Any],
    agent_identity_keys_by_id: Mapping[int, str] | None = None,
) -> dict[str, Any]:
    """Serialize an agent reference with an optional duplicate-name identity key."""
    entry: dict[str, Any] = {"name": agent.name}
    if agent_identity_keys_by_id is not None:
        identity = agent_identity_keys_by_id.get(id(agent))
        if identity is not None and identity != agent.name:
            entry["identity"] = identity
    return entry


def _ensure_json_compatible(value: Any) -> Any:
    try:
        return json.loads(json.dumps(value, default=str))
    except Exception:
        return str(value)


def _serialize_tool_call_data(tool_call: Any) -> Any:
    """Convert a tool call to a serializable dictionary."""
    return _serialize_raw_item_value(tool_call)


def _serialize_tool_metadata(
    tool: Any,
    *,
    include_description: bool = False,
    include_params_schema: bool = False,
) -> dict[str, Any]:
    """Build a dictionary of tool metadata for serialization."""
    metadata: dict[str, Any] = {"name": tool.name if hasattr(tool, "name") else None}
    namespace = get_function_tool_namespace(tool)
    if namespace is not None:
        metadata["namespace"] = namespace
    qualified_name = get_function_tool_qualified_name(tool)
    if qualified_name is not None and qualified_name != metadata["name"]:
        metadata["qualifiedName"] = qualified_name
    lookup_key = serialize_function_tool_lookup_key(get_function_tool_lookup_key_for_tool(tool))
    if lookup_key is not None:
        metadata["lookupKey"] = lookup_key
    if include_description and hasattr(tool, "description"):
        metadata["description"] = tool.description
    if include_params_schema and hasattr(tool, "params_json_schema"):
        metadata["paramsJsonSchema"] = tool.params_json_schema
    return metadata


def _serialize_tool_actions(
    actions: Sequence[Any],
    *,
    tool_attr: str,
    wrapper_key: str,
    include_description: bool = False,
    include_params_schema: bool = False,
) -> list[dict[str, Any]]:
    """Serialize tool action runs that share the same structure."""
    serialized_actions = []
    for action in actions:
        tool = getattr(action, tool_attr)
        tool_dict = _serialize_tool_metadata(
            tool,
            include_description=include_description,
            include_params_schema=include_params_schema,
        )
        serialized_actions.append(
            {
                "tool_call": _serialize_tool_call_data(action.tool_call),
                wrapper_key: tool_dict,
            }
        )
    return serialized_actions


def _serialize_handoffs(handoffs: Sequence[Any]) -> list[dict[str, Any]]:
    """Serialize handoff tool calls."""
    serialized_handoffs = []
    for handoff in handoffs:
        handoff_target = handoff.handoff
        handoff_name = _get_attr(handoff_target, "tool_name") or _get_attr(handoff_target, "name")
        serialized_handoffs.append(
            {
                "tool_call": _serialize_tool_call_data(handoff.tool_call),
                "handoff": {"tool_name": handoff_name},
            }
        )
    return serialized_handoffs


def _serialize_mcp_approval_requests(requests: Sequence[Any]) -> list[dict[str, Any]]:
    """Serialize MCP approval requests in a consistent format."""
    serialized_requests = []
    for request in requests:
        request_item_dict = _serialize_raw_item_value(request.request_item)
        serialized_requests.append(
            {
                "request_item": {"raw_item": request_item_dict},
                "mcp_tool": _serialize_mcp_tool(request.mcp_tool),
            }
        )
    return serialized_requests


def _serialize_mcp_tool(mcp_tool: Any) -> dict[str, Any]:
    """Serialize an MCP tool into a JSON-friendly mapping."""
    if mcp_tool is None:
        return {}

    tool_dict: dict[str, Any] | None = None
    if hasattr(mcp_tool, "to_json"):
        try:
            tool_json = mcp_tool.to_json()
        except Exception:
            tool_json = None
        if isinstance(tool_json, Mapping):
            tool_dict = dict(tool_json)
        elif tool_json is not None:
            tool_dict = {"value": tool_json}

    if tool_dict is None:
        tool_dict = _serialize_tool_metadata(mcp_tool)

    if tool_dict.get("name") is None:
        tool_dict["name"] = _get_attr(mcp_tool, "name")

    tool_config = _get_attr(mcp_tool, "tool_config")
    if tool_config is not None and "tool_config" not in tool_dict:
        tool_dict["tool_config"] = _serialize_raw_item_value(tool_config)

    normalized = _ensure_json_compatible(tool_dict)
    if isinstance(normalized, Mapping):
        return dict(normalized)
    return {"value": normalized}


def _serialize_tool_approval_interruption(
    interruption: ToolApprovalItem,
    *,
    include_tool_name: bool,
    agent_identity_keys_by_id: Mapping[int, str] | None = None,
) -> dict[str, Any]:
    """Serialize a ToolApprovalItem interruption."""
    interruption_dict: dict[str, Any] = {
        "type": "tool_approval_item",
        "raw_item": _serialize_raw_item_value(interruption.raw_item),
        "agent": _serialize_agent_reference(
            interruption.agent,
            agent_identity_keys_by_id=agent_identity_keys_by_id,
        ),
    }
    if include_tool_name and interruption.tool_name is not None:
        interruption_dict["tool_name"] = interruption.tool_name
    if interruption.tool_namespace is not None:
        interruption_dict["tool_namespace"] = interruption.tool_namespace
    if interruption.tool_origin is not None:
        interruption_dict["tool_origin"] = interruption.tool_origin.to_json_dict()
    tool_lookup_key = serialize_function_tool_lookup_key(
        getattr(interruption, "tool_lookup_key", None)
    )
    if tool_lookup_key is not None:
        interruption_dict["tool_lookup_key"] = tool_lookup_key
    if interruption._allow_bare_name_alias:
        interruption_dict["allow_bare_name_alias"] = True
    return interruption_dict


def _serialize_tool_action_groups(
    processed_response: ProcessedResponse,
) -> dict[str, list[dict[str, Any]]]:
    """Serialize tool-related action groups using a shared spec."""
    action_specs: list[
        tuple[str, list[Any], str, str, bool, bool]
    ] = [  # Key, actions, tool_attr, wrapper_key, include_description, include_params_schema.
        (
            "functions",
            processed_response.functions,
            "function_tool",
            "tool",
            True,
            True,
        ),
        (
            "computer_actions",
            processed_response.computer_actions,
            "computer_tool",
            "computer",
            True,
            False,
        ),
        (
            "custom_tool_actions",
            processed_response.custom_tool_calls,
            "custom_tool",
            "custom_tool",
            True,
            False,
        ),
        (
            "local_shell_actions",
            processed_response.local_shell_calls,
            "local_shell_tool",
            "local_shell",
            True,
            False,
        ),
        (
            "shell_actions",
            processed_response.shell_calls,
            "shell_tool",
            "shell",
            True,
            False,
        ),
        (
            "apply_patch_actions",
            processed_response.apply_patch_calls,
            "apply_patch_tool",
            "apply_patch",
            True,
            False,
        ),
    ]

    serialized: dict[str, list[dict[str, Any]]] = {
        key: _serialize_tool_actions(
            actions,
            tool_attr=tool_attr,
            wrapper_key=wrapper_key,
            include_description=include_description,
            include_params_schema=include_params_schema,
        )
        for (
            key,
            actions,
            tool_attr,
            wrapper_key,
            include_description,
            include_params_schema,
        ) in action_specs
    }
    serialized["handoffs"] = _serialize_handoffs(processed_response.handoffs)
    serialized["mcp_approval_requests"] = _serialize_mcp_approval_requests(
        processed_response.mcp_approval_requests
    )
    return serialized


def _serialize_pending_nested_agent_tool_runs(
    *,
    parent_state: RunState[Any, Any],
    function_entries: Sequence[dict[str, Any]],
    function_runs: Sequence[Any],
    scope_id: str | None = None,
    context_serializer: ContextSerializer | None = None,
    strict_context: bool = False,
    include_tracing_api_key: bool = False,
) -> None:
    """Attach serialized nested run state for pending agent-as-tool interruptions."""
    if not function_entries or not function_runs:
        return

    from .agent_tool_state import peek_agent_tool_run_result

    for entry, function_run in zip(function_entries, function_runs, strict=False):
        tool_call = getattr(function_run, "tool_call", None)
        if not isinstance(tool_call, ResponseFunctionToolCall):
            continue

        pending_run_result = peek_agent_tool_run_result(tool_call, scope_id=scope_id)
        if pending_run_result is None:
            continue

        interruptions = getattr(pending_run_result, "interruptions", None)
        if not isinstance(interruptions, list) or not interruptions:
            continue

        to_state = getattr(pending_run_result, "to_state", None)
        if not callable(to_state):
            continue

        try:
            nested_state = to_state()
        except Exception:
            if strict_context:
                raise
            logger.warning(
                "Failed to capture nested agent run state for tool call %s.",
                tool_call.call_id,
            )
            continue

        if not isinstance(nested_state, RunState):
            continue
        if nested_state is parent_state:
            # Defensive guard against accidental self-referential serialization loops.
            continue

        try:
            entry["agent_run_state"] = nested_state.to_json(
                context_serializer=context_serializer,
                strict_context=strict_context,
                include_tracing_api_key=include_tracing_api_key,
            )
        except Exception:
            if strict_context:
                raise
            logger.warning(
                "Failed to serialize nested agent run state for tool call %s.",
                tool_call.call_id,
            )


class _SerializedAgentToolRunResult:
    """Minimal run-result wrapper used to restore nested agent-as-tool resumptions."""

    def __init__(self, state: RunState[Any, Agent[Any]]) -> None:
        self._state = state
        self.interruptions = list(state.get_interruptions())
        self.final_output = None

    def to_state(self) -> RunState[Any, Agent[Any]]:
        return self._state


def _serialize_guardrail_results(
    results: Sequence[InputGuardrailResult | OutputGuardrailResult],
    *,
    agent_identity_keys_by_id: Mapping[int, str] | None = None,
) -> list[dict[str, Any]]:
    """Serialize guardrail results for persistence."""
    serialized: list[dict[str, Any]] = []
    for result in results:
        entry = {
            "guardrail": {
                "type": "output" if isinstance(result, OutputGuardrailResult) else "input",
                "name": result.guardrail.name,
            },
            "output": {
                "tripwireTriggered": result.output.tripwire_triggered,
                "outputInfo": result.output.output_info,
            },
        }
        if isinstance(result, OutputGuardrailResult):
            entry["agentOutput"] = result.agent_output
            entry["agent"] = _serialize_agent_reference(
                result.agent,
                agent_identity_keys_by_id=agent_identity_keys_by_id,
            )
        serialized.append(entry)
    return serialized


def _serialize_tool_guardrail_results(
    results: Sequence[ToolInputGuardrailResult | ToolOutputGuardrailResult],
    *,
    type_label: Literal["tool_input", "tool_output"],
) -> list[dict[str, Any]]:
    """Serialize tool guardrail results for persistence."""
    serialized: list[dict[str, Any]] = []
    for result in results:
        guardrail_name = (
            result.guardrail.get_name()
            if hasattr(result.guardrail, "get_name")
            else getattr(result.guardrail, "name", None)
        )
        serialized.append(
            {
                "guardrail": {"type": type_label, "name": guardrail_name},
                "output": {
                    "outputInfo": result.output.output_info,
                    "behavior": result.output.behavior,
                },
            }
        )
    return serialized


def _serialize_last_model_response(model_responses: list[dict[str, Any]]) -> Any:
    """Return the last serialized model response, if any."""
    if not model_responses:
        return None
    return model_responses[-1]


def _build_named_tool_map(
    tools: Sequence[Any], tool_type: type[Any]
) -> dict[NamedToolLookupKey, Any]:
    """Build a name-indexed map for tools of a given type."""
    if tool_type is FunctionTool:
        return cast(
            dict[NamedToolLookupKey, Any],
            build_function_tool_lookup_map(
                [tool for tool in tools if isinstance(tool, FunctionTool)]
            ),
        )

    tool_map: dict[NamedToolLookupKey, Any] = {}
    for tool in tools:
        if not isinstance(tool, tool_type) or not hasattr(tool, "name"):
            continue
        tool_name = getattr(tool, "name", None)
        if not isinstance(tool_name, str) or not tool_name:
            continue
        tool_map[tool_name] = tool
        if tool_type is ComputerTool:
            # Persisted runs may contain either the released preview name or the GA alias from
            # newer branches. Mirror both so either payload restores against the local tool.
            if tool_name == "computer":
                tool_map["computer_use_preview"] = tool
            elif tool_name == "computer_use_preview":
                tool_map["computer"] = tool
    return tool_map


def _build_handoffs_map(current_agent: Agent[Any]) -> dict[str, Handoff[Any, Agent[Any]]]:
    """Map handoff tool names to their definitions for quick lookup."""
    handoffs_map: dict[str, Handoff[Any, Agent[Any]]] = {}
    if not hasattr(current_agent, "handoffs"):
        return handoffs_map

    for handoff in current_agent.handoffs:
        if not isinstance(handoff, Handoff):
            continue
        handoff_name = getattr(handoff, "tool_name", None) or getattr(handoff, "name", None)
        if handoff_name:
            handoffs_map[handoff_name] = handoff
    return handoffs_map


async def _restore_pending_nested_agent_tool_runs(
    *,
    current_agent: Agent[Any],
    function_entries: Sequence[Any],
    function_runs: Sequence[Any],
    scope_id: str | None = None,
    context_deserializer: ContextDeserializer | None = None,
    strict_context: bool = False,
) -> None:
    """Rehydrate nested agent-as-tool run state into the ephemeral tool-call cache."""
    if not function_entries or not function_runs:
        return

    from .agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result

    for entry, function_run in zip(function_entries, function_runs, strict=False):
        if not isinstance(entry, Mapping):
            continue
        nested_state_data = entry.get("agent_run_state")
        if not isinstance(nested_state_data, Mapping):
            continue

        tool_call = getattr(function_run, "tool_call", None)
        if not isinstance(tool_call, ResponseFunctionToolCall):
            continue

        try:
            nested_state = await _build_run_state_from_json(
                initial_agent=current_agent,
                state_json=dict(nested_state_data),
                context_deserializer=context_deserializer,
                strict_context=strict_context,
            )
        except Exception:
            if strict_context:
                raise
            logger.warning(
                "Failed to deserialize nested agent run state for tool call %s.",
                tool_call.call_id,
            )
            continue

        pending_result = _SerializedAgentToolRunResult(nested_state)
        if not pending_result.interruptions:
            continue

        # Replace any stale cache entry with the same signature so resumed runs do not read
        # older pending interruptions after consuming this restored entry.
        drop_agent_tool_run_result(tool_call, scope_id=scope_id)
        record_agent_tool_run_result(tool_call, cast(Any, pending_result), scope_id=scope_id)


async def _deserialize_processed_response(
    processed_response_data: dict[str, Any],
    current_agent: Agent[Any],
    context: RunContextWrapper[Any],
    agent_map: dict[str, Agent[Any]],
    *,
    agent_identity_map: Mapping[str, Agent[Any]] | None = None,
    scope_id: str | None = None,
    context_deserializer: ContextDeserializer | None = None,
    strict_context: bool = False,
) -> ProcessedResponse:
    """Deserialize a ProcessedResponse from JSON data.

    Args:
        processed_response_data: Serialized ProcessedResponse dictionary.
        current_agent: The current agent (used to get tools and handoffs).
        context: The run context wrapper.
        agent_map: Map of agent names to agents.

    Returns:
        A reconstructed ProcessedResponse instance.
    """
    new_items = _deserialize_items(
        processed_response_data.get("new_items", []),
        agent_map,
        agent_identity_map=agent_identity_map,
    )

    if hasattr(current_agent, "get_all_tools"):
        all_tools = await current_agent.get_all_tools(context)
    else:
        all_tools = []

    tools_map = _build_named_tool_map(all_tools, FunctionTool)
    computer_tools_map = _build_named_tool_map(all_tools, ComputerTool)
    custom_tools_map = _build_named_tool_map(all_tools, CustomTool)
    local_shell_tools_map = _build_named_tool_map(all_tools, LocalShellTool)
    shell_tools_map = _build_named_tool_map(all_tools, ShellTool)
    apply_patch_tools_map = _build_named_tool_map(all_tools, ApplyPatchTool)
    mcp_tools_map = _build_named_tool_map(all_tools, HostedMCPTool)
    handoffs_map = _build_handoffs_map(current_agent)

    from .run_internal.run_steps import (
        ProcessedResponse,
        ToolRunApplyPatchCall,
        ToolRunComputerAction,
        ToolRunCustom,
        ToolRunFunction,
        ToolRunHandoff,
        ToolRunLocalShellCall,
        ToolRunMCPApprovalRequest,
        ToolRunShellCall,
    )

    def _deserialize_actions(
        entries: list[dict[str, Any]],
        *,
        tool_key: str,
        tool_map: Mapping[NamedToolLookupKey, Any],
        call_parser: Callable[[dict[str, Any]], Any],
        action_factory: Callable[[Any, Any], Any],
        name_resolver: Callable[[Mapping[str, Any]], NamedToolLookupKey | None] | None = None,
    ) -> list[Any]:
        """Deserialize tool actions with shared structure."""
        deserialized: list[Any] = []
        for entry in entries or []:
            tool_container = entry.get(tool_key, {}) if isinstance(entry, Mapping) else {}
            if name_resolver:
                tool_name = name_resolver(entry)
            else:
                if isinstance(tool_container, Mapping):
                    tool_name = tool_container.get("name")
                else:
                    tool_name = None
            tool = tool_map.get(tool_name) if tool_name else None
            if (
                tool is None
                and name_resolver is None
                and isinstance(tool_container, Mapping)
                and not isinstance(tool_container.get("namespace"), str)
            ):
                bare_name = tool_container.get("name")
                if isinstance(bare_name, str):
                    bare_lookup_key = get_function_tool_lookup_key(bare_name)
                    if bare_lookup_key is not None:
                        tool = tool_map.get(bare_lookup_key)
            if not tool:
                continue

            tool_call_data_raw = entry.get("tool_call", {}) if isinstance(entry, Mapping) else {}
            tool_call_data = (
                dict(tool_call_data_raw) if isinstance(tool_call_data_raw, Mapping) else {}
            )
            try:
                tool_call = call_parser(tool_call_data)
            except Exception:
                continue
            deserialized.append(action_factory(tool_call, tool))
        return deserialized

    def _parse_with_adapter(adapter: TypeAdapter[Any], data: dict[str, Any]) -> Any:
        try:
            return adapter.validate_python(data)
        except ValidationError:
            return data

    def _parse_apply_patch_call(data: dict[str, Any]) -> Any:
        try:
            return ResponseFunctionToolCall(**data)
        except Exception:
            return data

    def _deserialize_action_groups() -> dict[str, list[Any]]:
        def _resolve_handoff_tool_name(data: Mapping[str, Any]) -> NamedToolLookupKey | None:
            handoff_data = data.get("handoff", {})
            if not isinstance(handoff_data, Mapping):
                return None
            tool_name = handoff_data.get("tool_name")
            return cast(
                NamedToolLookupKey | None, tool_name if isinstance(tool_name, str) else None
            )

        def _resolve_function_tool_name(data: Mapping[str, Any]) -> FunctionToolLookupKey | None:
            tool_data = data.get("tool", {})
            if isinstance(tool_data, Mapping):
                lookup_key = deserialize_function_tool_lookup_key(tool_data.get("lookupKey"))
                if lookup_key is not None:
                    return lookup_key

            tool_call_data = data.get("tool_call", {})
            if isinstance(tool_call_data, Mapping):
                lookup_key = get_function_tool_lookup_key(
                    cast(str | None, tool_call_data.get("name")),
                    cast(str | None, tool_call_data.get("namespace")),
                )
                if lookup_key is not None:
                    return lookup_key

            if not isinstance(tool_data, Mapping):
                return None
            return get_function_tool_lookup_key(
                cast(str | None, tool_data.get("name")),
                cast(str | None, tool_data.get("namespace")),
            )

        action_specs: list[
            tuple[
                str,
                str,
                Mapping[Any, Any],
                Callable[[dict[str, Any]], Any],
                Callable[[Any, Any], Any],
                Callable[[Mapping[str, Any]], NamedToolLookupKey | None] | None,
            ]
        ] = [
            (
                "handoffs",
                "handoff",
                handoffs_map,
                lambda data: ResponseFunctionToolCall(**data),
                lambda tool_call, handoff: ToolRunHandoff(tool_call=tool_call, handoff=handoff),
                _resolve_handoff_tool_name,
            ),
            (
                "functions",
                "tool",
                tools_map,
                lambda data: ResponseFunctionToolCall(**data),
                lambda tool_call, function_tool: ToolRunFunction(
                    tool_call=tool_call, function_tool=function_tool
                ),
                _resolve_function_tool_name,
            ),
            (
                "computer_actions",
                "computer",
                computer_tools_map,
                lambda data: ResponseComputerToolCall(**data),
                lambda tool_call, computer_tool: ToolRunComputerAction(
                    tool_call=tool_call, computer_tool=computer_tool
                ),
                None,
            ),
            (
                "custom_tool_actions",
                "custom_tool",
                custom_tools_map,
                lambda data: ResponseCustomToolCall(**data),
                lambda tool_call, custom_tool: ToolRunCustom(
                    tool_call=tool_call, custom_tool=custom_tool
                ),
                None,
            ),
            (
                "local_shell_actions",
                "local_shell",
                local_shell_tools_map,
                lambda data: _parse_with_adapter(_LOCAL_SHELL_CALL_ADAPTER, data),
                lambda tool_call, local_shell_tool: ToolRunLocalShellCall(
                    tool_call=tool_call, local_shell_tool=local_shell_tool
                ),
                None,
            ),
            (
                "shell_actions",
                "shell",
                shell_tools_map,
                lambda data: _parse_with_adapter(_LOCAL_SHELL_CALL_ADAPTER, data),
                lambda tool_call, shell_tool: ToolRunShellCall(
                    tool_call=tool_call, shell_tool=shell_tool
                ),
                None,
            ),
            (
                "apply_patch_actions",
                "apply_patch",
                apply_patch_tools_map,
                _parse_apply_patch_call,
                lambda tool_call, apply_patch_tool: ToolRunApplyPatchCall(
                    tool_call=tool_call, apply_patch_tool=apply_patch_tool
                ),
                None,
            ),
        ]

        action_groups: dict[str, list[Any]] = {}
        for (
            key,
            tool_key,
            tool_map,
            call_parser,
            action_factory,
            name_resolver,
        ) in action_specs:
            action_groups[key] = _deserialize_actions(
                processed_response_data.get(key, []),
                tool_key=tool_key,
                tool_map=tool_map,
                call_parser=call_parser,
                action_factory=action_factory,
                name_resolver=name_resolver,
            )
        return action_groups

    action_groups = _deserialize_action_groups()
    handoffs = action_groups["handoffs"]
    functions = action_groups["functions"]
    computer_actions = action_groups["computer_actions"]
    custom_tool_actions = action_groups["custom_tool_actions"]
    local_shell_actions = action_groups["local_shell_actions"]
    shell_actions = action_groups["shell_actions"]
    apply_patch_actions = action_groups["apply_patch_actions"]

    await _restore_pending_nested_agent_tool_runs(
        current_agent=current_agent,
        function_entries=processed_response_data.get("functions", []),
        function_runs=functions,
        scope_id=scope_id,
        context_deserializer=context_deserializer,
        strict_context=strict_context,
    )

    mcp_approval_requests: list[ToolRunMCPApprovalRequest] = []
    for request_data in processed_response_data.get("mcp_approval_requests", []):
        request_item_data = request_data.get("request_item", {})
        raw_item_data = (
            request_item_data.get("raw_item", {}) if isinstance(request_item_data, Mapping) else {}
        )
        request_item_adapter: TypeAdapter[McpApprovalRequest] = TypeAdapter(McpApprovalRequest)
        request_item = request_item_adapter.validate_python(raw_item_data)

        mcp_tool_data = request_data.get("mcp_tool", {})
        if not mcp_tool_data:
            continue

        mcp_tool_name = mcp_tool_data.get("name")
        mcp_tool = mcp_tools_map.get(mcp_tool_name) if mcp_tool_name else None

        if mcp_tool:
            mcp_approval_requests.append(
                ToolRunMCPApprovalRequest(
                    request_item=request_item,
                    mcp_tool=mcp_tool,
                )
            )

    interruptions: list[ToolApprovalItem] = []
    for interruption_data in processed_response_data.get("interruptions", []):
        approval_item = _deserialize_tool_approval_item(
            interruption_data,
            agent_map=agent_map,
            agent_identity_map=agent_identity_map,
            fallback_agent=current_agent,
        )
        if approval_item is not None:
            interruptions.append(approval_item)

    return ProcessedResponse(
        new_items=new_items,
        handoffs=handoffs,
        functions=functions,
        computer_actions=computer_actions,
        custom_tool_calls=custom_tool_actions,
        local_shell_calls=local_shell_actions,
        shell_calls=shell_actions,
        apply_patch_calls=apply_patch_actions,
        tools_used=processed_response_data.get("tools_used", []),
        mcp_approval_requests=mcp_approval_requests,
        interruptions=interruptions,
    )


def _deserialize_tool_call_raw_item(normalized_raw_item: Mapping[str, Any]) -> Any:
    """Deserialize a tool call raw item when possible, falling back to the original mapping."""
    if not isinstance(normalized_raw_item, Mapping):
        return normalized_raw_item

    tool_type = normalized_raw_item.get("type")

    if tool_type == "function_call":
        try:
            return ResponseFunctionToolCall(**normalized_raw_item)
        except Exception:
            return normalized_raw_item

    if tool_type in {"shell_call", "apply_patch_call", "hosted_tool_call", "local_shell_call"}:
        return normalized_raw_item

    try:
        return ResponseFunctionToolCall(**normalized_raw_item)
    except Exception:
        return normalized_raw_item


def _can_construct_statusless_message(exc: ValidationError) -> bool:
    missing_fields = {
        str(error["loc"][0])
        for error in exc.errors()
        if error.get("type") == "missing"
        and isinstance(error.get("loc"), tuple)
        and error.get("loc")
    }
    if not missing_fields:
        return False
    return missing_fields <= _ALLOWED_MISSING_MESSAGE_FIELDS


def _deserialize_message_content_part(value: object) -> object:
    if not isinstance(value, Mapping):
        return value

    part_type = value.get("type")
    if part_type == "output_text":
        return ResponseOutputText.model_construct(**dict(value))
    if part_type == "refusal":
        return ResponseOutputRefusal.model_construct(**dict(value))
    return dict(value)


def _deserialize_message_output_item(payload: Mapping[str, Any]) -> ResponseOutputMessage:
    try:
        return ResponseOutputMessage(**payload)
    except ValidationError as exc:
        if not _can_construct_statusless_message(exc):
            raise

    content = payload.get("content")
    normalized_content = (
        [_deserialize_message_content_part(part) for part in content]
        if isinstance(content, list)
        else content
    )
    normalized_payload = dict(payload)
    normalized_payload["content"] = normalized_content
    return ResponseOutputMessage.model_construct(**normalized_payload)


def _resolve_agent_from_data(
    agent_data: Any,
    agent_map: Mapping[str, Agent[Any]],
    agent_identity_map: Mapping[str, Agent[Any]] | None = None,
    fallback_agent: Agent[Any] | None = None,
) -> Agent[Any] | None:
    """Resolve an agent from serialized data with an optional fallback."""
    agent_name = None
    agent_identity = None
    if isinstance(agent_data, Mapping):
        agent_identity = agent_data.get("identity")
        agent_name = agent_data.get("name")
    elif isinstance(agent_data, str):
        agent_name = agent_data

    if isinstance(agent_identity, str) and agent_identity_map is not None:
        resolved = agent_identity_map.get(agent_identity)
        if resolved is not None:
            return resolved
        raise UserError(
            "Run state references an agent identity that is not present in the restored graph: "
            f"{agent_identity}"
        )

    if agent_name:
        if agent_identity_map is not None:
            resolved = agent_identity_map.get(agent_name)
            if resolved is not None:
                return resolved
        return agent_map.get(agent_name) or fallback_agent
    return fallback_agent


def _deserialize_tool_approval_raw_item(normalized_raw_item: Any) -> Any:
    """Deserialize a tool approval raw item, preferring function calls when possible."""
    if not isinstance(normalized_raw_item, Mapping):
        return normalized_raw_item

    return _deserialize_tool_call_raw_item(dict(normalized_raw_item))


def _deserialize_tool_approval_item(
    item_data: Mapping[str, Any],
    *,
    agent_map: Mapping[str, Agent[Any]],
    agent_identity_map: Mapping[str, Agent[Any]] | None = None,
    fallback_agent: Agent[Any] | None = None,
    pre_normalized_raw_item: Any | None = None,
) -> ToolApprovalItem | None:
    """Deserialize a ToolApprovalItem from serialized data."""
    agent = _resolve_agent_from_data(
        item_data.get("agent"),
        agent_map,
        agent_identity_map,
        fallback_agent,
    )
    if agent is None:
        return None

    raw_item_data: Any = pre_normalized_raw_item
    if raw_item_data is None:
        raw_item_data = item_data.get("raw_item") or item_data.get("rawItem") or {}
        if isinstance(raw_item_data, Mapping):
            raw_item_data = dict(raw_item_data)

    tool_name = item_data.get("tool_name")
    tool_namespace = item_data.get("tool_namespace")
    tool_origin = _deserialize_tool_origin(item_data.get("tool_origin"))
    tool_lookup_key = deserialize_function_tool_lookup_key(item_data.get("tool_lookup_key"))
    allow_bare_name_alias = item_data.get("allow_bare_name_alias") is True
    raw_item = _deserialize_tool_approval_raw_item(raw_item_data)
    return ToolApprovalItem(
        agent=agent,
        raw_item=raw_item,
        tool_name=tool_name,
        tool_namespace=tool_namespace,
        tool_origin=tool_origin,
        tool_lookup_key=tool_lookup_key,
        _allow_bare_name_alias=allow_bare_name_alias,
    )


def _deserialize_tool_call_output_raw_item(
    raw_item: Mapping[str, Any],
) -> FunctionCallOutput | ComputerCallOutput | LocalShellCallOutput | dict[str, Any] | None:
    """Deserialize a tool call output raw item; return None when validation fails."""
    if not isinstance(raw_item, Mapping):
        return cast(
            FunctionCallOutput | ComputerCallOutput | LocalShellCallOutput | dict[str, Any],
            raw_item,
        )

    normalized_raw_item = dict(raw_item)
    output_type = normalized_raw_item.get("type")

    if output_type == "function_call_output":
        return _FUNCTION_OUTPUT_ADAPTER.validate_python(normalized_raw_item)
    if output_type == "computer_call_output":
        return _COMPUTER_OUTPUT_ADAPTER.validate_python(normalized_raw_item)
    if output_type == "local_shell_call_output":
        return _LOCAL_SHELL_OUTPUT_ADAPTER.validate_python(normalized_raw_item)
    if output_type in {"shell_call_output", "apply_patch_call_output", "custom_tool_call_output"}:
        return normalized_raw_item

    try:
        return cast(
            FunctionCallOutput | ComputerCallOutput | LocalShellCallOutput | dict[str, Any],
            _TOOL_CALL_OUTPUT_UNION_ADAPTER.validate_python(normalized_raw_item),
        )
    except ValidationError:
        return None


def _parse_guardrail_entry(
    entry: Any, *, expected_type: Literal["input", "output"]
) -> tuple[str, GuardrailFunctionOutput, dict[str, Any]] | None:
    entry_dict = entry if isinstance(entry, dict) else {}
    guardrail_info_raw = entry_dict.get("guardrail", {})
    guardrail_info = guardrail_info_raw if isinstance(guardrail_info_raw, dict) else {}
    guardrail_type = guardrail_info.get("type")
    if guardrail_type and guardrail_type != expected_type:
        return None
    name = guardrail_info.get("name") or f"deserialized_{expected_type}_guardrail"
    output_data_raw = entry_dict.get("output", {})
    output_data = output_data_raw if isinstance(output_data_raw, dict) else {}
    guardrail_output = GuardrailFunctionOutput(
        output_info=output_data.get("outputInfo"),
        tripwire_triggered=bool(output_data.get("tripwireTriggered")),
    )
    return name, guardrail_output, entry_dict


def _parse_tool_guardrail_entry(
    entry: Any, *, expected_type: Literal["tool_input", "tool_output"]
) -> tuple[str, ToolGuardrailFunctionOutput] | None:
    entry_dict = entry if isinstance(entry, dict) else {}
    guardrail_info_raw = entry_dict.get("guardrail", {})
    guardrail_info = guardrail_info_raw if isinstance(guardrail_info_raw, dict) else {}
    guardrail_type = guardrail_info.get("type")
    if guardrail_type and guardrail_type != expected_type:
        return None
    name = guardrail_info.get("name") or f"deserialized_{expected_type}_guardrail"
    output_data_raw = entry_dict.get("output", {})
    output_data = output_data_raw if isinstance(output_data_raw, dict) else {}
    behavior_data = output_data.get("behavior")
    behavior: RejectContentBehavior | RaiseExceptionBehavior | AllowBehavior
    if isinstance(behavior_data, dict) and "type" in behavior_data:
        behavior = cast(
            RejectContentBehavior | RaiseExceptionBehavior | AllowBehavior,
            behavior_data,
        )
    else:
        behavior = AllowBehavior(type="allow")
    output_info = output_data.get("outputInfo")
    guardrail_output = ToolGuardrailFunctionOutput(
        output_info=output_info,
        behavior=behavior,
    )
    return name, guardrail_output


def _deserialize_input_guardrail_results(
    results_data: list[dict[str, Any]],
) -> list[InputGuardrailResult]:
    """Rehydrate input guardrail results from serialized data."""
    deserialized: list[InputGuardrailResult] = []
    for entry in results_data or []:
        parsed = _parse_guardrail_entry(entry, expected_type="input")
        if not parsed:
            continue
        name, guardrail_output, _ = parsed

        def _input_guardrail_fn(
            context: RunContextWrapper[Any],
            agent: Agent[Any],
            input: Any,
            *,
            _output: GuardrailFunctionOutput = guardrail_output,
        ) -> GuardrailFunctionOutput:
            return _output

        guardrail = InputGuardrail(guardrail_function=_input_guardrail_fn, name=name)
        deserialized.append(InputGuardrailResult(guardrail=guardrail, output=guardrail_output))
    return deserialized


def _deserialize_output_guardrail_results(
    results_data: list[dict[str, Any]],
    *,
    agent_map: dict[str, Agent[Any]],
    agent_identity_map: Mapping[str, Agent[Any]] | None = None,
    fallback_agent: Agent[Any],
) -> list[OutputGuardrailResult]:
    """Rehydrate output guardrail results from serialized data."""
    deserialized: list[OutputGuardrailResult] = []
    for entry in results_data or []:
        parsed = _parse_guardrail_entry(entry, expected_type="output")
        if not parsed:
            continue
        name, guardrail_output, entry_dict = parsed
        agent_output = entry_dict.get("agentOutput")
        agent_data = entry_dict.get("agent")
        resolved_agent = _resolve_agent_from_data(
            agent_data,
            agent_map,
            agent_identity_map,
            fallback_agent,
        )
        if resolved_agent is None:
            resolved_agent = fallback_agent

        def _output_guardrail_fn(
            context: RunContextWrapper[Any],
            agent_param: Agent[Any],
            agent_output_param: Any,
            *,
            _output: GuardrailFunctionOutput = guardrail_output,
        ) -> GuardrailFunctionOutput:
            return _output

        guardrail = OutputGuardrail(guardrail_function=_output_guardrail_fn, name=name)
        deserialized.append(
            OutputGuardrailResult(
                guardrail=guardrail,
                agent_output=agent_output,
                agent=resolved_agent,
                output=guardrail_output,
            )
        )
    return deserialized


def _deserialize_tool_input_guardrail_results(
    results_data: list[dict[str, Any]],
) -> list[ToolInputGuardrailResult]:
    """Rehydrate tool input guardrail results from serialized data."""
    deserialized: list[ToolInputGuardrailResult] = []
    for entry in results_data or []:
        parsed = _parse_tool_guardrail_entry(entry, expected_type="tool_input")
        if not parsed:
            continue
        name, guardrail_output = parsed

        def _tool_input_guardrail_fn(
            data: Any,
            *,
            _output: ToolGuardrailFunctionOutput = guardrail_output,
        ) -> ToolGuardrailFunctionOutput:
            return _output

        guardrail: ToolInputGuardrail[Any] = ToolInputGuardrail(
            guardrail_function=_tool_input_guardrail_fn, name=name
        )
        deserialized.append(ToolInputGuardrailResult(guardrail=guardrail, output=guardrail_output))
    return deserialized


def _deserialize_tool_output_guardrail_results(
    results_data: list[dict[str, Any]],
) -> list[ToolOutputGuardrailResult]:
    """Rehydrate tool output guardrail results from serialized data."""
    deserialized: list[ToolOutputGuardrailResult] = []
    for entry in results_data or []:
        parsed = _parse_tool_guardrail_entry(entry, expected_type="tool_output")
        if not parsed:
            continue
        name, guardrail_output = parsed

        def _tool_output_guardrail_fn(
            data: Any,
            *,
            _output: ToolGuardrailFunctionOutput = guardrail_output,
        ) -> ToolGuardrailFunctionOutput:
            return _output

        guardrail: ToolOutputGuardrail[Any] = ToolOutputGuardrail(
            guardrail_function=_tool_output_guardrail_fn, name=name
        )
        deserialized.append(ToolOutputGuardrailResult(guardrail=guardrail, output=guardrail_output))
    return deserialized


async def _build_run_state_from_json(
    initial_agent: Agent[Any],
    state_json: dict[str, Any],
    context_override: ContextOverride | None = None,
    context_deserializer: ContextDeserializer | None = None,
    strict_context: bool = False,
) -> RunState[Any, Agent[Any]]:
    """Shared helper to rebuild RunState from JSON payload.

    Context restoration follows this precedence order:

    1. ``context_override`` when supplied.
    2. ``context_deserializer`` applied to serialized mapping data.
    3. Direct mapping restore for contexts that were serialized as plain mappings.

    When the snapshot metadata indicates that the original context type could not round-trip
    safely, this function warns or raises (in ``strict_context`` mode) rather than silently
    claiming that the rebuilt mapping is equivalent to the original object.
    """
    schema_version = state_json.get("$schemaVersion")
    if not schema_version:
        raise UserError("Run state is missing schema version")
    if schema_version not in SUPPORTED_SCHEMA_VERSIONS:
        supported_versions = ", ".join(sorted(SUPPORTED_SCHEMA_VERSIONS))
        raise UserError(
            f"Run state schema version {schema_version} is not supported. "
            f"Supported versions are: {supported_versions}. "
            f"New snapshots are written as version {CURRENT_SCHEMA_VERSION}."
        )

    agent_identity_map = _build_agent_identity_map(initial_agent)
    agent_map = _build_agent_map(initial_agent)

    current_agent_data = state_json["current_agent"]
    current_agent_name = current_agent_data["name"]
    current_agent = _resolve_agent_from_data(
        current_agent_data,
        agent_map,
        agent_identity_map=agent_identity_map,
    )
    if not current_agent:
        raise UserError(f"Agent {current_agent_name} not found in agent map")

    context_data = state_json["context"]
    usage = deserialize_usage(context_data.get("usage", {}))

    serialized_context: Any = context_data.get("context", _MISSING_CONTEXT_SENTINEL)
    if serialized_context is _MISSING_CONTEXT_SENTINEL:
        serialized_context = {}
    context_meta_raw = context_data.get("context_meta")
    context_meta = context_meta_raw if isinstance(context_meta_raw, Mapping) else None

    # If context was originally a custom type and no override/deserializer is supplied,
    # surface the risk of losing behavior/state during restore.
    if (
        context_override is None
        and context_deserializer is None
        and _context_meta_requires_deserializer(context_meta)
    ):
        warning_message = _context_meta_warning_message(context_meta)
        if strict_context:
            raise UserError(warning_message)
        logger.warning(warning_message)

    if isinstance(context_override, RunContextWrapper):
        context = context_override
    elif context_override is not None:
        context = RunContextWrapper(context=context_override)
    elif serialized_context is None:
        context = RunContextWrapper(context=None)
    elif context_deserializer is not None:
        if not isinstance(serialized_context, Mapping):
            raise UserError(
                "Serialized run state context must be a mapping to use context_deserializer."
            )
        try:
            rebuilt_context = context_deserializer(dict(serialized_context))
        except Exception as exc:
            raise UserError(
                "Context deserializer failed while rebuilding RunState context."
            ) from exc
        if isinstance(rebuilt_context, RunContextWrapper):
            context = rebuilt_context
        else:
            context = RunContextWrapper(context=rebuilt_context)
    elif isinstance(serialized_context, Mapping):
        context = RunContextWrapper(context=serialized_context)
    else:
        raise UserError("Serialized run state context must be a mapping. Please provide one.")
    context.usage = usage
    context._rebuild_approvals(context_data.get("approvals", {}))
    serialized_tool_input = context_data.get("tool_input")
    if (
        context_override is None
        and serialized_tool_input is not None
        and getattr(context, "tool_input", None) is None
    ):
        context.tool_input = serialized_tool_input

    original_input_raw = state_json["original_input"]
    if isinstance(original_input_raw, list):
        normalized_original_input = []
        for item in original_input_raw:
            if not isinstance(item, Mapping):
                normalized_original_input.append(item)
                continue
            item_dict = dict(item)
            normalized_original_input.append(item_dict)
    else:
        normalized_original_input = original_input_raw

    state = RunState(
        context=context,
        original_input=normalized_original_input,
        starting_agent=current_agent,
        max_turns=state_json["max_turns"],
        conversation_id=state_json.get("conversation_id"),
        previous_response_id=state_json.get("previous_response_id"),
        auto_previous_response_id=bool(state_json.get("auto_previous_response_id", False)),
    )
    state._starting_agent = initial_agent
    state._schema_version = schema_version
    from .agent_tool_state import set_agent_tool_state_scope

    state._agent_tool_state_scope_id = uuid4().hex
    set_agent_tool_state_scope(context, state._agent_tool_state_scope_id)

    state._current_turn = state_json["current_turn"]
    state._model_responses = _deserialize_model_responses(state_json.get("model_responses", []))
    state._generated_items = _deserialize_items(
        state_json.get("generated_items", []),
        agent_map,
        agent_identity_map=agent_identity_map,
    )

    last_processed_response_data = state_json.get("last_processed_response")
    if last_processed_response_data and state._context is not None:
        state._last_processed_response = await _deserialize_processed_response(
            last_processed_response_data,
            current_agent,
            state._context,
            agent_map,
            agent_identity_map=agent_identity_map,
            scope_id=state._agent_tool_state_scope_id,
            context_deserializer=context_deserializer,
            strict_context=strict_context,
        )
    else:
        state._last_processed_response = None

    if "session_items" in state_json:
        state._session_items = _deserialize_items(
            state_json.get("session_items", []),
            agent_map,
            agent_identity_map=agent_identity_map,
        )
    else:
        state._session_items = state._merge_generated_items_with_processed()

    state._mark_generated_items_merged_with_last_processed()

    state._input_guardrail_results = _deserialize_input_guardrail_results(
        state_json.get("input_guardrail_results", [])
    )
    state._output_guardrail_results = _deserialize_output_guardrail_results(
        state_json.get("output_guardrail_results", []),
        agent_map=agent_map,
        agent_identity_map=agent_identity_map,
        fallback_agent=current_agent,
    )
    state._tool_input_guardrail_results = _deserialize_tool_input_guardrail_results(
        state_json.get("tool_input_guardrail_results", [])
    )
    state._tool_output_guardrail_results = _deserialize_tool_output_guardrail_results(
        state_json.get("tool_output_guardrail_results", [])
    )

    current_step_data = state_json.get("current_step")
    if current_step_data and current_step_data.get("type") == "next_step_interruption":
        interruptions: list[ToolApprovalItem] = []
        interruptions_data = current_step_data.get("data", {}).get(
            "interruptions", current_step_data.get("interruptions", [])
        )
        for item_data in interruptions_data:
            approval_item = _deserialize_tool_approval_item(
                item_data,
                agent_map=agent_map,
                agent_identity_map=agent_identity_map,
            )
            if approval_item is not None:
                interruptions.append(approval_item)

        from .run_internal.run_steps import NextStepInterruption

        state._current_step = NextStepInterruption(
            interruptions=[item for item in interruptions if isinstance(item, ToolApprovalItem)]
        )

    state._current_turn_persisted_item_count = state_json.get(
        "current_turn_persisted_item_count", 0
    )
    serialized_policy = state_json.get("reasoning_item_id_policy")
    if serialized_policy in {"preserve", "omit"}:
        state._reasoning_item_id_policy = cast(Literal["preserve", "omit"], serialized_policy)
    else:
        state._reasoning_item_id_policy = None
    serialized_prompt_cache_key = state_json.get("generated_prompt_cache_key")
    state._generated_prompt_cache_key = (
        serialized_prompt_cache_key if isinstance(serialized_prompt_cache_key, str) else None
    )
    state.set_tool_use_tracker_snapshot(state_json.get("tool_use_tracker", {}))
    trace_data = state_json.get("trace")
    if isinstance(trace_data, Mapping):
        state._trace_state = TraceState.from_json(trace_data)
    else:
        state._trace_state = None
    sandbox_data = state_json.get("sandbox")
    state._sandbox = dict(sandbox_data) if isinstance(sandbox_data, Mapping) else None

    return state


def _iter_agent_graph(initial_agent: Agent[Any]) -> Iterator[Agent[Any]]:
    """Yield agents reachable from the starting agent in breadth-first order."""
    queue: deque[Agent[Any]] = deque([initial_agent])
    seen_agent_ids: set[int] = set()

    while queue:
        current = queue.popleft()
        current_id = id(current)
        if current_id in seen_agent_ids:
            continue
        seen_agent_ids.add(current_id)
        yield current

        for handoff_item in current.handoffs:
            handoff_agent: Any | None = None
            handoff_agent_name: str | None = None

            if isinstance(handoff_item, Handoff):
                # Some custom/mocked Handoff subclasses bypass dataclass initialization.
                # Prefer agent_name, then legacy name fallback used in tests.
                candidate_name = getattr(handoff_item, "agent_name", None) or getattr(
                    handoff_item, "name", None
                )
                if isinstance(candidate_name, str):
                    handoff_agent_name = candidate_name

                handoff_ref = getattr(handoff_item, "_agent_ref", None)
                handoff_agent = handoff_ref() if callable(handoff_ref) else None
                if handoff_agent is None:
                    # Backward-compatibility fallback for custom legacy handoff objects that store
                    # the target directly on `.agent`. New code should prefer `handoff()` objects.
                    legacy_agent = getattr(handoff_item, "agent", None)
                    if legacy_agent is not None:
                        handoff_agent = legacy_agent
                        logger.debug(
                            "Using legacy handoff `.agent` fallback while building agent map. "
                            "This compatibility path is not recommended for new code."
                        )
                if handoff_agent_name is None:
                    candidate_name = getattr(handoff_agent, "name", None)
                    handoff_agent_name = candidate_name if isinstance(candidate_name, str) else None
                if handoff_agent is None or not hasattr(handoff_agent, "handoffs"):
                    if handoff_agent_name:
                        logger.debug(
                            "Skipping unresolved handoff target while building agent map: %s",
                            handoff_agent_name,
                        )
                    continue
            else:
                # Backward-compatibility fallback for custom legacy handoff wrappers that expose
                # the target directly on `.agent` without inheriting from `Handoff`.
                legacy_agent = getattr(handoff_item, "agent", None)
                if legacy_agent is not None:
                    handoff_agent = legacy_agent
                    logger.debug(
                        "Using legacy non-`Handoff` `.agent` fallback while building agent map."
                    )
                else:
                    handoff_agent = handoff_item
                candidate_name = getattr(handoff_agent, "name", None)
                handoff_agent_name = candidate_name if isinstance(candidate_name, str) else None

            if handoff_agent is not None and handoff_agent_name:
                queue.append(cast(Agent[Any], handoff_agent))

        # Include agent-as-tool instances so nested approvals can be restored.
        tools = getattr(current, "tools", None)
        if tools:
            for tool in tools:
                if not getattr(tool, "_is_agent_tool", False):
                    continue
                tool_agent = getattr(tool, "_agent_instance", None)
                tool_agent_name = getattr(tool_agent, "name", None)
                if tool_agent and tool_agent_name:
                    queue.append(tool_agent)


def _allocate_unique_agent_identity(agent_name: str, used_identities: set[str]) -> str:
    """Return a deterministic identity key without colliding with literal agent names."""
    candidate = agent_name
    next_index = 1
    while candidate in used_identities:
        next_index += 1
        candidate = f"{agent_name}#{next_index}"
    used_identities.add(candidate)
    return candidate


def _identity_type_name(value: Any) -> str:
    return f"{type(value).__module__}.{type(value).__qualname__}"


def _callable_identity_name(value: Any) -> str:
    module = getattr(value, "__module__", type(value).__module__)
    qualname = getattr(value, "__qualname__", type(value).__qualname__)
    return f"{module}.{qualname}"


def _normalize_identity_value(value: Any) -> Any:
    if value is None or isinstance(value, str | int | float | bool):
        return value
    if isinstance(value, bytes | bytearray):
        return {"type": "bytes", "length": len(value)}
    if callable(value):
        return {"callable": _callable_identity_name(value)}
    if dataclasses.is_dataclass(value):
        return {
            "dataclass": _identity_type_name(value),
            "value": _normalize_identity_value(dataclasses.asdict(cast(Any, value))),
        }
    if hasattr(value, "model_dump"):
        try:
            dumped = value.model_dump(exclude_unset=True)
        except TypeError:
            dumped = value.model_dump()
        return {
            "model": _identity_type_name(value),
            "value": _normalize_identity_value(dumped),
        }
    if isinstance(value, Mapping):
        return {
            str(key): _normalize_identity_value(item)
            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
        }
    if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray):
        return [_normalize_identity_value(item) for item in value]

    value_name = getattr(value, "name", None)
    if isinstance(value_name, str):
        return {"type": _identity_type_name(value), "name": value_name}
    return {"type": _identity_type_name(value)}


def _stable_identity_text(value: Any) -> str:
    return json.dumps(
        _normalize_identity_value(value),
        sort_keys=True,
        separators=(",", ":"),
    )


def _tool_identity_signature(tool: Any) -> dict[str, Any]:
    signature: dict[str, Any] = {
        "type": _identity_type_name(tool),
        "name": getattr(tool, "name", None),
    }
    namespace = get_function_tool_namespace(tool)
    if namespace is not None:
        signature["namespace"] = namespace
    qualified_name = get_function_tool_qualified_name(tool)
    if qualified_name is not None:
        signature["qualified_name"] = qualified_name
    if hasattr(tool, "environment"):
        signature["environment"] = _normalize_identity_value(tool.environment)
    if getattr(tool, "_is_agent_tool", False):
        nested_agent = getattr(tool, "_agent_instance", None)
        signature["agent_tool_target"] = getattr(nested_agent, "name", None)
    return signature


_THREADING_LOCK_TYPES = (type(threading.Lock()), type(threading.RLock()))


def _is_capability_runtime_only_value(value: Any) -> bool:
    return isinstance(
        value,
        (
            BaseSandboxSession,
            asyncio.Event,
            asyncio.Lock,
            asyncio.Semaphore,
            asyncio.Condition,
            threading.Event,
            *_THREADING_LOCK_TYPES,
        ),
    )


def _normalize_capability_identity_value(
    value: Any,
    *,
    seen: set[int] | None = None,
) -> Any:
    if seen is None:
        seen = set()

    if value is None or isinstance(value, str | int | float | bool):
        return value
    if isinstance(value, Path):
        return value.as_posix()
    if isinstance(value, bytes | bytearray):
        return {"type": "bytes", "length": len(value)}
    if callable(value):
        return {"callable": _callable_identity_name(value)}
    if _is_capability_runtime_only_value(value):
        return {"runtime_only": _identity_type_name(value)}
    if isinstance(
        value,
        ApplyPatchTool | ComputerTool | FunctionTool | HostedMCPTool | LocalShellTool | ShellTool,
    ):
        return _tool_identity_signature(value)

    object_id = id(value)
    if object_id in seen:
        return {"recursive": _identity_type_name(value)}

    if dataclasses.is_dataclass(value):
        seen.add(object_id)
        try:
            merged_fields = {
                field.name: getattr(value, field.name) for field in dataclasses.fields(value)
            }
            if hasattr(value, "__dict__"):
                for name, item in vars(value).items():
                    if name.startswith("_") or name in merged_fields:
                        continue
                    merged_fields[name] = item
            return {
                "dataclass": _identity_type_name(value),
                "value": {
                    name: _normalize_capability_identity_value(
                        item,
                        seen=seen,
                    )
                    for name, item in sorted(merged_fields.items())
                },
            }
        finally:
            seen.remove(object_id)

    if isinstance(value, Capability):
        seen.add(object_id)
        try:
            merged_fields = {}
            for name, field_info in value.__class__.model_fields.items():
                if field_info.exclude or name.startswith("_") or name == "session":
                    continue
                merged_fields[name] = getattr(value, name)
            return {
                "capability": _identity_type_name(value),
                "value": {
                    name: _normalize_capability_identity_value(
                        item,
                        seen=seen,
                    )
                    for name, item in sorted(merged_fields.items())
                },
            }
        finally:
            seen.remove(object_id)

    if hasattr(value, "model_dump"):
        seen.add(object_id)
        try:
            try:
                dumped = value.model_dump(mode="json", round_trip=True)
            except TypeError:
                dumped = value.model_dump(mode="json")
            return {
                "model": _identity_type_name(value),
                "value": _normalize_capability_identity_value(dumped, seen=seen),
            }
        finally:
            seen.remove(object_id)

    if isinstance(value, Mapping):
        seen.add(object_id)
        try:
            return {
                str(key): _normalize_capability_identity_value(item, seen=seen)
                for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
            }
        finally:
            seen.remove(object_id)

    if isinstance(value, set | frozenset):
        seen.add(object_id)
        try:
            normalized_items = [
                _normalize_capability_identity_value(item, seen=seen) for item in value
            ]
            return sorted(normalized_items, key=_stable_identity_text)
        finally:
            seen.remove(object_id)

    if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray):
        seen.add(object_id)
        try:
            return [_normalize_capability_identity_value(item, seen=seen) for item in value]
        finally:
            seen.remove(object_id)

    if hasattr(value, "__dict__"):
        seen.add(object_id)
        try:
            return {
                "object": _identity_type_name(value),
                "value": {
                    name: _normalize_capability_identity_value(item, seen=seen)
                    for name, item in sorted(vars(value).items())
                    if not name.startswith("_")
                },
            }
        finally:
            seen.remove(object_id)

    value_name = getattr(value, "name", None)
    if isinstance(value_name, str):
        return {"type": _identity_type_name(value), "name": value_name}
    return {"type": _identity_type_name(value)}


def _capability_identity_signature(capability: Any) -> dict[str, Any]:
    return {
        "type": _identity_type_name(capability),
        "value": _normalize_capability_identity_value(capability),
    }


def _handoff_identity_signature(handoff_item: Agent[Any] | Handoff[Any, Any]) -> dict[str, Any]:
    if isinstance(handoff_item, Handoff):
        tool_name = getattr(handoff_item, "tool_name", None)
        if not isinstance(tool_name, str):
            tool_name = getattr(handoff_item, "name", None)
        agent_name = getattr(handoff_item, "agent_name", None)
        return {
            "type": _identity_type_name(handoff_item),
            "tool_name": tool_name,
            "agent_name": agent_name if isinstance(agent_name, str) else None,
            "input_filter": _normalize_identity_value(getattr(handoff_item, "input_filter", None)),
            "nest_handoff_history": getattr(handoff_item, "nest_handoff_history", None),
        }

    return {
        "type": _identity_type_name(handoff_item),
        "agent_name": getattr(handoff_item, "name", None),
    }


def _agent_identity_signature(agent: Agent[Any]) -> str:
    signature: dict[str, Any] = {
        "agent_type": _identity_type_name(agent),
        "handoff_description": getattr(agent, "handoff_description", None),
        "instructions": _normalize_identity_value(getattr(agent, "instructions", None)),
        "prompt": _normalize_identity_value(getattr(agent, "prompt", None)),
        "model": _normalize_identity_value(getattr(agent, "model", None)),
        "model_settings": _normalize_identity_value(getattr(agent, "model_settings", None)),
        "mcp_config": _normalize_capability_identity_value(getattr(agent, "mcp_config", None)),
        "hooks": _normalize_capability_identity_value(getattr(agent, "hooks", None)),
        "input_guardrails": sorted(
            _stable_identity_text(_normalize_capability_identity_value(guardrail))
            for guardrail in getattr(agent, "input_guardrails", [])
        ),
        "output_guardrails": sorted(
            _stable_identity_text(_normalize_capability_identity_value(guardrail))
            for guardrail in getattr(agent, "output_guardrails", [])
        ),
        "output_type": _normalize_identity_value(getattr(agent, "output_type", None)),
        "tool_use_behavior": _normalize_capability_identity_value(
            getattr(agent, "tool_use_behavior", None)
        ),
        "reset_tool_choice": getattr(agent, "reset_tool_choice", None),
        "tools": sorted(
            _stable_identity_text(_tool_identity_signature(tool))
            for tool in getattr(agent, "tools", [])
        ),
        "handoffs": sorted(
            _stable_identity_text(_handoff_identity_signature(handoff_item))
            for handoff_item in getattr(agent, "handoffs", [])
        ),
        "mcp_servers": sorted(
            _stable_identity_text(server) for server in getattr(agent, "mcp_servers", [])
        ),
    }

    default_manifest = getattr(agent, "default_manifest", None)
    if default_manifest is not None:
        signature["default_manifest"] = _normalize_capability_identity_value(default_manifest)

    base_instructions = getattr(agent, "base_instructions", None)
    if base_instructions is not None:
        signature["base_instructions"] = _normalize_identity_value(base_instructions)

    capabilities = getattr(agent, "capabilities", None)
    if isinstance(capabilities, Sequence):
        signature["capabilities"] = sorted(
            _stable_identity_text(_capability_identity_signature(capability))
            for capability in capabilities
        )

    return _stable_identity_text(signature)


def _agent_identity_sort_key(
    agent: Agent[Any],
    *,
    root_agent: Agent[Any],
    original_index: int,
) -> tuple[int, str, int]:
    return (
        0 if agent is root_agent else 1,
        _agent_identity_signature(agent),
        original_index,
    )


def _build_agent_identity_map(initial_agent: Agent[Any]) -> dict[str, Agent[Any]]:
    """Build a stable identity map that preserves duplicate agent names."""
    ordered_agents = list(_iter_agent_graph(initial_agent))
    original_indices = {id(agent): index for index, agent in enumerate(ordered_agents)}
    literal_names = {agent.name for agent in ordered_agents}
    agents_by_name: dict[str, list[Agent[Any]]] = {}
    for agent in ordered_agents:
        agents_by_name.setdefault(agent.name, []).append(agent)

    agent_identity_map: dict[str, Agent[Any]] = {}
    used_identities: set[str] = set()
    processed_names: set[str] = set()

    for agent in ordered_agents:
        agent_name = agent.name
        if agent_name in processed_names:
            continue
        processed_names.add(agent_name)

        group = agents_by_name[agent_name]
        sorted_group = sorted(
            group,
            key=lambda candidate: _agent_identity_sort_key(
                candidate,
                root_agent=initial_agent,
                original_index=original_indices[id(candidate)],
            ),
        )

        base_agent = sorted_group[0]
        used_identities.add(agent_name)
        agent_identity_map[agent_name] = base_agent

        next_index = 2
        for duplicate_agent in sorted_group[1:]:
            candidate = f"{agent_name}#{next_index}"
            while candidate in used_identities or candidate in literal_names:
                next_index += 1
                candidate = f"{agent_name}#{next_index}"
            used_identities.add(candidate)
            agent_identity_map[candidate] = duplicate_agent
            next_index += 1

    return agent_identity_map


def _build_agent_identity_keys_by_id(initial_agent: Agent[Any]) -> dict[int, str]:
    """Build stable identity keys for the reachable agent graph."""
    return {
        id(agent): identity for identity, agent in _build_agent_identity_map(initial_agent).items()
    }


def _build_agent_map(initial_agent: Agent[Any]) -> dict[str, Agent[Any]]:
    """Build a map of agent names to agents by traversing handoffs.

    Args:
        initial_agent: The starting agent.

    Returns:
        Dictionary mapping agent names to agent instances.
    """
    agent_map: dict[str, Agent[Any]] = {}
    for agent in _iter_agent_graph(initial_agent):
        agent_map.setdefault(agent.name, agent)

    return agent_map


def _deserialize_model_responses(responses_data: list[dict[str, Any]]) -> list[ModelResponse]:
    """Deserialize model responses from JSON data.

    Args:
        responses_data: List of serialized model response dictionaries.

    Returns:
        List of ModelResponse instances.
    """

    result = []
    for resp_data in responses_data:
        usage = deserialize_usage(resp_data.get("usage", {}))

        output: list[Any] = [
            _deserialize_message_output_item(item)
            if isinstance(item, Mapping) and item.get("type") == "message"
            else item
            for item in resp_data["output"]
        ]

        response_id = resp_data.get("response_id")
        request_id = resp_data.get("request_id")

        result.append(
            ModelResponse(
                usage=usage,
                output=output,
                response_id=response_id,
                request_id=request_id,
            )
        )

    return result


def _deserialize_items(
    items_data: list[dict[str, Any]],
    agent_map: dict[str, Agent[Any]],
    *,
    agent_identity_map: Mapping[str, Agent[Any]] | None = None,
) -> list[RunItem]:
    """Deserialize run items from JSON data.

    Args:
        items_data: List of serialized run item dictionaries.
        agent_map: Map of agent names to agent instances.

    Returns:
        List of RunItem instances.
    """

    result: list[RunItem] = []

    def _resolve_agent_info(
        item_data: Mapping[str, Any], item_type: str
    ) -> tuple[Agent[Any] | None, str | None]:
        """Resolve agent from serialized data."""
        candidate_name: str | None = None
        fields = ["agent"]
        if item_type == "handoff_output_item":
            fields.extend(["source_agent", "target_agent"])

        for agent_field in fields:
            raw_agent = item_data.get(agent_field)
            if isinstance(raw_agent, Mapping):
                candidate_name = raw_agent.get("name") or candidate_name
            elif isinstance(raw_agent, str):
                candidate_name = raw_agent

            agent_candidate = _resolve_agent_from_data(
                raw_agent,
                agent_map,
                agent_identity_map,
            )
            if agent_candidate:
                return agent_candidate, agent_candidate.name

        return None, candidate_name

    for item_data in items_data:
        item_type = item_data.get("type")
        if not item_type:
            logger.warning("Item missing type field, skipping")
            continue

        agent, agent_name = _resolve_agent_info(item_data, item_type)
        if not agent:
            if agent_name:
                logger.warning(f"Agent {agent_name} not found, skipping item")
            else:
                logger.warning(f"Item missing agent field, skipping: {item_type}")
            continue

        raw_item_data = item_data["raw_item"]
        normalized_raw_item = (
            dict(raw_item_data) if isinstance(raw_item_data, Mapping) else raw_item_data
        )

        try:
            if item_type == "message_output_item":
                raw_item_msg = _deserialize_message_output_item(normalized_raw_item)
                result.append(MessageOutputItem(agent=agent, raw_item=raw_item_msg))

            elif item_type == "tool_search_call_item":
                raw_item_tool_search_call = coerce_tool_search_call_raw_item(normalized_raw_item)
                result.append(ToolSearchCallItem(agent=agent, raw_item=raw_item_tool_search_call))

            elif item_type == "tool_search_output_item":
                raw_item_tool_search_output = coerce_tool_search_output_raw_item(
                    normalized_raw_item
                )
                result.append(
                    ToolSearchOutputItem(agent=agent, raw_item=raw_item_tool_search_output)
                )

            elif item_type == "tool_call_item":
                # Tool call items can be function calls, shell calls, apply_patch calls,
                # MCP calls, etc. Check the type field to determine which type to deserialize as
                raw_item_tool = _deserialize_tool_call_raw_item(normalized_raw_item)
                # Preserve display metadata if it was stored with the item.
                description = item_data.get("description")
                title = item_data.get("title")
                tool_origin = _deserialize_tool_origin(item_data.get("tool_origin"))
                result.append(
                    ToolCallItem(
                        agent=agent,
                        raw_item=raw_item_tool,
                        description=description,
                        title=title,
                        tool_origin=tool_origin,
                    )
                )

            elif item_type == "tool_call_output_item":
                # For tool call outputs, validate and convert the raw dict
                # Try to determine the type based on the dict structure
                raw_item_output = _deserialize_tool_call_output_raw_item(normalized_raw_item)
                if raw_item_output is None:
                    continue
                result.append(
                    ToolCallOutputItem(
                        agent=agent,
                        raw_item=raw_item_output,
                        output=item_data.get("output", ""),
                        tool_origin=_deserialize_tool_origin(item_data.get("tool_origin")),
                    )
                )

            elif item_type == "reasoning_item":
                raw_item_reason = ResponseReasoningItem(**normalized_raw_item)
                result.append(ReasoningItem(agent=agent, raw_item=raw_item_reason))

            elif item_type == "handoff_call_item":
                raw_item_handoff = ResponseFunctionToolCall(**normalized_raw_item)
                result.append(HandoffCallItem(agent=agent, raw_item=raw_item_handoff))

            elif item_type == "handoff_output_item":
                source_agent = _resolve_agent_from_data(
                    item_data.get("source_agent"),
                    agent_map,
                    agent_identity_map,
                )
                target_agent = _resolve_agent_from_data(
                    item_data.get("target_agent"),
                    agent_map,
                    agent_identity_map,
                )

                # If we cannot resolve both agents, skip this item gracefully
                if not source_agent or not target_agent:
                    source_name = item_data.get("source_agent")
                    target_name = item_data.get("target_agent")
                    logger.warning(
                        "Skipping handoff_output_item: could not resolve agents "
                        "(source=%s, target=%s).",
                        source_name,
                        target_name,
                    )
                    continue

                # For handoff output items, we need to validate the raw_item
                # as a TResponseInputItem (which is a union type)
                # If validation fails, use the raw dict as-is (for test compatibility)
                try:
                    raw_item_handoff_output = _HANDOFF_OUTPUT_ADAPTER.validate_python(
                        normalized_raw_item
                    )
                except ValidationError:
                    # If validation fails, use the raw dict as-is
                    # This allows tests to use mock data that doesn't match
                    # the exact TResponseInputItem union types
                    raw_item_handoff_output = normalized_raw_item  # type: ignore[assignment]
                result.append(
                    HandoffOutputItem(
                        agent=agent,
                        raw_item=raw_item_handoff_output,
                        source_agent=source_agent,
                        target_agent=target_agent,
                    )
                )

            elif item_type == "compaction_item":
                try:
                    raw_item_compaction = _HANDOFF_OUTPUT_ADAPTER.validate_python(
                        normalized_raw_item
                    )
                except ValidationError:
                    raw_item_compaction = normalized_raw_item  # type: ignore[assignment]
                result.append(CompactionItem(agent=agent, raw_item=raw_item_compaction))

            elif item_type == "mcp_list_tools_item":
                raw_item_mcp_list = McpListTools(**normalized_raw_item)
                result.append(MCPListToolsItem(agent=agent, raw_item=raw_item_mcp_list))

            elif item_type == "mcp_approval_request_item":
                raw_item_mcp_req = McpApprovalRequest(**normalized_raw_item)
                result.append(MCPApprovalRequestItem(agent=agent, raw_item=raw_item_mcp_req))

            elif item_type == "mcp_approval_response_item":
                # Validate and convert the raw dict to McpApprovalResponse
                raw_item_mcp_response = _MCP_APPROVAL_RESPONSE_ADAPTER.validate_python(
                    normalized_raw_item
                )
                result.append(MCPApprovalResponseItem(agent=agent, raw_item=raw_item_mcp_response))

            elif item_type == "tool_approval_item":
                approval_item = _deserialize_tool_approval_item(
                    item_data,
                    agent_map=agent_map,
                    agent_identity_map=agent_identity_map,
                    fallback_agent=agent,
                    pre_normalized_raw_item=normalized_raw_item,
                )
                if approval_item is not None:
                    result.append(approval_item)

        except UserError:
            raise
        except Exception as e:
            logger.warning(f"Failed to deserialize item of type {item_type}: {e}")
            continue

    return result


def _clone_original_input(original_input: str | list[Any]) -> str | list[Any]:
    """Return a deep copy of the original input so later mutations don't leak into saved state."""
    if isinstance(original_input, str):
        return original_input
    return copy.deepcopy(original_input)