repo_description: A lightweight, powerful framework for multi-agent workflows
repo_topics: agents ai framework llm openai python
code: """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)
  