items.py

"""
Item utilities for the run pipeline. Hosts input normalization helpers and lightweight builders
for synthetic run items or IDs used during tool execution. Internal use only.
"""

from __future__ import annotations

import json
from collections.abc import Sequence
from typing import Any, Literal, cast

from openai.types.responses import ResponseFunctionToolCall
from pydantic import BaseModel

from ..agent_tool_state import drop_agent_tool_run_result
from ..items import ItemHelpers, RunItem, ToolCallOutputItem, TResponseInputItem
from ..models.fake_id import FAKE_RESPONSES_ID
from ..tool import DEFAULT_APPROVAL_REJECTION_MESSAGE

REJECTION_MESSAGE = DEFAULT_APPROVAL_REJECTION_MESSAGE
TOOL_CALL_SESSION_DESCRIPTION_KEY = "_agents_tool_description"
TOOL_CALL_SESSION_TITLE_KEY = "_agents_tool_title"
_TOOL_CALL_TO_OUTPUT_TYPE: dict[str, str] = {
    "function_call": "function_call_output",
    "shell_call": "shell_call_output",
    "apply_patch_call": "apply_patch_call_output",
    "computer_call": "computer_call_output",
    "local_shell_call": "local_shell_call_output",
    "tool_search_call": "tool_search_output",
}

__all__ = [
    "ReasoningItemIdPolicy",
    "REJECTION_MESSAGE",
    "TOOL_CALL_SESSION_DESCRIPTION_KEY",
    "TOOL_CALL_SESSION_TITLE_KEY",
    "copy_input_items",
    "drop_orphan_function_calls",
    "ensure_input_item_format",
    "prepare_model_input_items",
    "run_item_to_input_item",
    "run_items_to_input_items",
    "normalize_input_items_for_api",
    "normalize_resumed_input",
    "fingerprint_input_item",
    "deduplicate_input_items",
    "deduplicate_input_items_preferring_latest",
    "strip_internal_input_item_metadata",
    "function_rejection_item",
    "shell_rejection_item",
    "apply_patch_rejection_item",
    "extract_mcp_request_id",
    "extract_mcp_request_id_from_run",
]


ReasoningItemIdPolicy = Literal["preserve", "omit"]


def copy_input_items(value: str | list[TResponseInputItem]) -> str | list[TResponseInputItem]:
    """Return a shallow copy of input items so mutations do not leak between turns."""
    return value if isinstance(value, str) else value.copy()


def run_item_to_input_item(
    run_item: RunItem,
    reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
) -> TResponseInputItem | None:
    """Convert a run item to model input, optionally stripping reasoning IDs."""
    if run_item.type == "tool_approval_item":
        return None
    to_input = getattr(run_item, "to_input_item", None)
    input_item = to_input() if callable(to_input) else cast(TResponseInputItem, run_item.raw_item)
    if (
        _should_omit_reasoning_item_ids(reasoning_item_id_policy)
        and run_item.type == "reasoning_item"
    ):
        return _without_reasoning_item_id(input_item)
    return cast(TResponseInputItem, input_item)


def run_items_to_input_items(
    run_items: Sequence[RunItem],
    reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
) -> list[TResponseInputItem]:
    """Convert run items to model input items while skipping approvals."""
    converted: list[TResponseInputItem] = []
    for run_item in run_items:
        item = run_item_to_input_item(run_item, reasoning_item_id_policy)
        if item is not None:
            converted.append(item)
    return converted


def drop_orphan_function_calls(
    items: list[TResponseInputItem],
    *,
    pruning_indexes: set[int] | None = None,
) -> list[TResponseInputItem]:
    """
    Remove tool call items that do not have corresponding outputs so resumptions or retries do not
    replay stale tool calls.
    """

    completed_call_ids = _completed_call_ids_by_type(items)
    matched_anonymous_tool_search_calls = _matched_anonymous_tool_search_call_indexes(items)

    filtered: list[TResponseInputItem] = []
    for index, entry in enumerate(items):
        if not isinstance(entry, dict):
            filtered.append(entry)
            continue
        entry_type = entry.get("type")
        if not isinstance(entry_type, str):
            filtered.append(entry)
            continue
        output_type = _TOOL_CALL_TO_OUTPUT_TYPE.get(entry_type)
        if output_type is None:
            filtered.append(entry)
            continue
        if pruning_indexes is not None and index not in pruning_indexes:
            filtered.append(entry)
            continue
        call_id = entry.get("call_id")
        if isinstance(call_id, str) and call_id in completed_call_ids.get(output_type, set()):
            filtered.append(entry)
            continue
        if (
            entry_type == "tool_search_call"
            and not isinstance(call_id, str)
            and index in matched_anonymous_tool_search_calls
        ):
            filtered.append(entry)
    return filtered


def ensure_input_item_format(item: TResponseInputItem) -> TResponseInputItem:
    """Ensure a single item is normalized for model input."""
    coerced = _coerce_to_dict(item)
    if coerced is None:
        return item

    return cast(TResponseInputItem, coerced)


def normalize_input_items_for_api(items: list[TResponseInputItem]) -> list[TResponseInputItem]:
    """Normalize input items for API submission."""

    normalized: list[TResponseInputItem] = []
    for item in items:
        coerced = _coerce_to_dict(item)
        if coerced is None:
            normalized.append(item)
            continue

        normalized_item = strip_internal_input_item_metadata(cast(TResponseInputItem, coerced))
        normalized.append(normalized_item)
    return normalized


def prepare_model_input_items(
    caller_items: Sequence[TResponseInputItem],
    generated_items: Sequence[TResponseInputItem] = (),
) -> list[TResponseInputItem]:
    """Normalize model input while pruning orphans only from runner-generated history."""
    normalized_caller_items = normalize_input_items_for_api(list(caller_items))
    if not generated_items:
        return normalized_caller_items

    normalized_generated_items = normalize_input_items_for_api(list(generated_items))
    filtered_generated_items = drop_orphan_function_calls(normalized_generated_items)
    return normalized_caller_items + filtered_generated_items


def normalize_resumed_input(
    raw_input: str | list[TResponseInputItem],
) -> str | list[TResponseInputItem]:
    """Normalize resumed list inputs and drop orphan tool calls."""
    if isinstance(raw_input, list):
        normalized = normalize_input_items_for_api(raw_input)
        return drop_orphan_function_calls(normalized)
    return raw_input


def fingerprint_input_item(item: Any, *, ignore_ids_for_matching: bool = False) -> str | None:
    """Hashable fingerprint used to dedupe or rewind input items across resumes."""
    if item is None:
        return None

    try:
        payload: Any
        if hasattr(item, "model_dump"):
            payload = _model_dump_without_warnings(item)
            if payload is None:
                return None
            if isinstance(payload, dict):
                payload = cast(
                    dict[str, Any],
                    strip_internal_input_item_metadata(cast(TResponseInputItem, payload)),
                )
        elif isinstance(item, dict):
            payload = cast(
                dict[str, Any],
                strip_internal_input_item_metadata(cast(TResponseInputItem, item)),
            )
            if ignore_ids_for_matching:
                payload.pop("id", None)
        else:
            payload = ensure_input_item_format(item)
            if isinstance(payload, dict):
                payload = cast(
                    dict[str, Any],
                    strip_internal_input_item_metadata(cast(TResponseInputItem, payload)),
                )
            if ignore_ids_for_matching and isinstance(payload, dict):
                payload.pop("id", None)

        return json.dumps(payload, sort_keys=True, default=str)
    except Exception:
        return None


def _dedupe_key(item: TResponseInputItem) -> str | None:
    """Return a stable identity key when items carry explicit identifiers."""
    payload = _coerce_to_dict(item)
    if payload is None:
        return None

    role = payload.get("role")
    item_type = payload.get("type") or role
    if role is not None or item_type == "message":
        return None
    item_id = payload.get("id")
    if item_id == FAKE_RESPONSES_ID:
        # Ignore placeholder IDs so call_id-based dedupe remains possible.
        item_id = None
    if isinstance(item_id, str):
        return f"id:{item_type}:{item_id}"

    call_id = payload.get("call_id")
    if isinstance(call_id, str):
        return f"call_id:{item_type}:{call_id}"

    # points back to the originating approval request ID on hosted MCP responses
    approval_request_id = payload.get("approval_request_id")
    if isinstance(approval_request_id, str):
        return f"approval_request_id:{item_type}:{approval_request_id}"

    return None


def strip_internal_input_item_metadata(item: TResponseInputItem) -> TResponseInputItem:
    """Remove SDK-only session metadata before sending items back to the model."""
    if not isinstance(item, dict):
        return item

    cleaned = dict(item)
    cleaned.pop(TOOL_CALL_SESSION_DESCRIPTION_KEY, None)
    cleaned.pop(TOOL_CALL_SESSION_TITLE_KEY, None)
    return cast(TResponseInputItem, cleaned)


def _should_omit_reasoning_item_ids(reasoning_item_id_policy: ReasoningItemIdPolicy | None) -> bool:
    return reasoning_item_id_policy == "omit"


def _without_reasoning_item_id(item: TResponseInputItem) -> TResponseInputItem:
    if not isinstance(item, dict):
        return item
    if item.get("type") != "reasoning":
        return item
    if "id" not in item:
        return item
    sanitized = dict(item)
    sanitized.pop("id", None)
    return cast(TResponseInputItem, sanitized)


def deduplicate_input_items(items: Sequence[TResponseInputItem]) -> list[TResponseInputItem]:
    """Remove duplicate items that share stable identifiers to avoid re-sending tool outputs."""
    seen_keys: set[str] = set()
    deduplicated: list[TResponseInputItem] = []
    for item in items:
        dedupe_key = _dedupe_key(item)
        if dedupe_key is None:
            deduplicated.append(item)
            continue
        if dedupe_key in seen_keys:
            continue
        seen_keys.add(dedupe_key)
        deduplicated.append(item)
    return deduplicated


def deduplicate_input_items_preferring_latest(
    items: Sequence[TResponseInputItem],
) -> list[TResponseInputItem]:
    """Deduplicate by stable identifiers while keeping the latest occurrence."""
    # deduplicate_input_items keeps the first item per dedupe key. Reverse twice so that
    # the latest item in the original order wins for duplicate IDs/call_ids.
    return list(reversed(deduplicate_input_items(list(reversed(items)))))


def function_rejection_item(
    agent: Any,
    tool_call: Any,
    *,
    rejection_message: str = REJECTION_MESSAGE,
    scope_id: str | None = None,
    tool_origin: Any = None,
) -> ToolCallOutputItem:
    """Build a ToolCallOutputItem representing a rejected function tool call."""
    if isinstance(tool_call, ResponseFunctionToolCall):
        drop_agent_tool_run_result(tool_call, scope_id=scope_id)
    return ToolCallOutputItem(
        output=rejection_message,
        raw_item=ItemHelpers.tool_call_output_item(tool_call, rejection_message),
        agent=agent,
        tool_origin=tool_origin,
    )


def shell_rejection_item(
    agent: Any,
    call_id: str,
    *,
    rejection_message: str = REJECTION_MESSAGE,
) -> ToolCallOutputItem:
    """Build a ToolCallOutputItem representing a rejected shell call."""
    rejection_output: dict[str, Any] = {
        "stdout": "",
        "stderr": rejection_message,
        "outcome": {"type": "exit", "exit_code": 1},
    }
    rejection_raw_item: dict[str, Any] = {
        "type": "shell_call_output",
        "call_id": call_id,
        "output": [rejection_output],
    }
    return ToolCallOutputItem(agent=agent, output=rejection_message, raw_item=rejection_raw_item)


def apply_patch_rejection_item(
    agent: Any,
    call_id: str,
    *,
    rejection_message: str = REJECTION_MESSAGE,
) -> ToolCallOutputItem:
    """Build a ToolCallOutputItem representing a rejected apply_patch call."""
    rejection_raw_item: dict[str, Any] = {
        "type": "apply_patch_call_output",
        "call_id": call_id,
        "status": "failed",
        "output": rejection_message,
    }
    return ToolCallOutputItem(
        agent=agent,
        output=rejection_message,
        raw_item=rejection_raw_item,
    )


def extract_mcp_request_id(raw_item: Any) -> str | None:
    """Pull the request id from hosted MCP approval payloads."""
    if isinstance(raw_item, dict):
        provider_data = raw_item.get("provider_data")
        if isinstance(provider_data, dict):
            candidate = provider_data.get("id")
            if isinstance(candidate, str):
                return candidate
        candidate = raw_item.get("id") or raw_item.get("call_id")
        return candidate if isinstance(candidate, str) else None
    try:
        provider_data = getattr(raw_item, "provider_data", None)
    except Exception:
        provider_data = None
    if isinstance(provider_data, dict):
        candidate = provider_data.get("id")
        if isinstance(candidate, str):
            return candidate
    try:
        candidate = getattr(raw_item, "id", None) or getattr(raw_item, "call_id", None)
    except Exception:
        candidate = None
    return candidate if isinstance(candidate, str) else None


def extract_mcp_request_id_from_run(mcp_run: Any) -> str | None:
    """Extract the hosted MCP request id from a streaming run item."""
    request_item = getattr(mcp_run, "request_item", None) or getattr(mcp_run, "requestItem", None)
    if isinstance(request_item, dict):
        provider_data = request_item.get("provider_data")
        if isinstance(provider_data, dict):
            candidate = provider_data.get("id")
            if isinstance(candidate, str):
                return candidate
        candidate = request_item.get("id") or request_item.get("call_id")
    else:
        provider_data = getattr(request_item, "provider_data", None)
        if isinstance(provider_data, dict):
            candidate = provider_data.get("id")
            if isinstance(candidate, str):
                return candidate
        candidate = getattr(request_item, "id", None) or getattr(request_item, "call_id", None)
    return candidate if isinstance(candidate, str) else None


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


def _completed_call_ids_by_type(payload: list[TResponseInputItem]) -> dict[str, set[str]]:
    """Return call ids that already have outputs, grouped by output type."""
    completed: dict[str, set[str]] = {
        output_type: set() for output_type in _TOOL_CALL_TO_OUTPUT_TYPE.values()
    }
    for entry in payload:
        if not isinstance(entry, dict):
            continue
        item_type = entry.get("type")
        if not isinstance(item_type, str) or item_type not in completed:
            continue
        call_id = entry.get("call_id")
        if isinstance(call_id, str):
            completed[item_type].add(call_id)
    return completed


def _matched_anonymous_tool_search_call_indexes(payload: list[TResponseInputItem]) -> set[int]:
    """Return anonymous tool_search_call indexes that have a later anonymous output."""
    matched_indexes: set[int] = set()
    pending_anonymous_outputs = 0

    for index in range(len(payload) - 1, -1, -1):
        entry = payload[index]
        if not isinstance(entry, dict):
            continue

        item_type = entry.get("type")
        if item_type == "tool_search_output" and not isinstance(entry.get("call_id"), str):
            pending_anonymous_outputs += 1
            continue

        if (
            item_type == "tool_search_call"
            and not isinstance(entry.get("call_id"), str)
            and pending_anonymous_outputs > 0
        ):
            matched_indexes.add(index)
            pending_anonymous_outputs -= 1

    return matched_indexes


def _coerce_to_dict(value: object) -> dict[str, Any] | None:
    """Convert model items to dicts so fields can be renamed and sanitized."""
    if isinstance(value, dict):
        return dict(value)
    if isinstance(value, BaseModel):
        return _model_dump_without_warnings(value)
    if hasattr(value, "model_dump"):
        return _model_dump_without_warnings(value)
    return None


def _model_dump_without_warnings(value: object) -> dict[str, Any] | None:
    """Best-effort model_dump that avoids noisy serialization warnings from third-party models."""
    if not hasattr(value, "model_dump"):
        return None

    model_dump = cast(Any, value).model_dump
    try:
        return cast(dict[str, Any], model_dump(exclude_unset=True, warnings=False))
    except TypeError:
        # Some model_dump-compatible objects only accept exclude_unset.
        try:
            return cast(dict[str, Any], model_dump(exclude_unset=True))
        except Exception:
            return None
    except Exception:
        return None