diff --git a/src/agents/handoffs/__init__.py b/src/agents/handoffs/__init__.py index 106f46e6ce..c1902e0b47 100644 --- a/src/agents/handoffs/__init__.py +++ b/src/agents/handoffs/__init__.py @@ -80,7 +80,11 @@ def clone(self, **kwargs: Any) -> HandoffInputData: ``` """ - return dataclasses_replace(self, **kwargs) + cloned = dataclasses_replace(self, **kwargs) + owned_items = getattr(self, "_nested_history_owned_items", ()) + if owned_items: + object.__setattr__(cloned, "_nested_history_owned_items", owned_items) + return cloned HandoffInputFilter: TypeAlias = Callable[[HandoffInputData], MaybeAwaitable[HandoffInputData]] diff --git a/src/agents/handoffs/history.py b/src/agents/handoffs/history.py index f734f0ab9b..00e83b4cc1 100644 --- a/src/agents/handoffs/history.py +++ b/src/agents/handoffs/history.py @@ -2,6 +2,7 @@ import json from copy import deepcopy +from dataclasses import replace from typing import TYPE_CHECKING, Any, cast from ..items import ( @@ -12,6 +13,7 @@ ) if TYPE_CHECKING: + from ..run_internal.items import NestedHistoryOwnedItem from . import HandoffHistoryMapper, HandoffInputData __all__ = [ @@ -83,42 +85,211 @@ def nest_handoff_history( ) -> HandoffInputData: """Summarize the previous transcript for the next agent.""" + nested, _ = _nest_handoff_history_with_provenance( + handoff_input_data, + history_mapper=history_mapper, + ) + return nested + + +def _nest_handoff_history_with_provenance( + handoff_input_data: HandoffInputData, + *, + history_mapper: HandoffHistoryMapper | None = None, +) -> tuple[HandoffInputData, tuple[NestedHistoryOwnedItem, ...]]: + """Return nested input and exact provenance for items moved into default history.""" + normalized_history = _normalize_input_history(handoff_input_data.input_history) - flattened_history = _flatten_nested_history_messages(normalized_history) + flattened_history = [ + _strip_transcript_item_metadata(item) + for item in _flatten_nested_history_messages(normalized_history) + ] - # Convert items to plain inputs for the transcript summary. - pre_items_as_inputs: list[TResponseInputItem] = [] - filtered_pre_items: list[RunItem] = [] + # Partition items between summary segments and lossless model input while retaining order. + normalized_pre_items: list[tuple[TResponseInputItem, bool, RunItem]] = [] for run_item in handoff_input_data.pre_handoff_items: if isinstance(run_item, ToolApprovalItem): continue plain_input = _run_item_to_plain_input(run_item) - pre_items_as_inputs.append(plain_input) - if _should_forward_pre_item(plain_input): - filtered_pre_items.append(run_item) + forward_verbatim = _should_forward_pre_item(plain_input) + normalized_pre_items.append((plain_input, forward_verbatim, run_item)) - new_items_as_inputs: list[TResponseInputItem] = [] - filtered_input_items: list[RunItem] = [] + normalized_new_items: list[tuple[TResponseInputItem, bool, RunItem]] = [] for run_item in handoff_input_data.new_items: if isinstance(run_item, ToolApprovalItem): continue plain_input = _run_item_to_plain_input(run_item) - new_items_as_inputs.append(plain_input) - if _should_forward_new_item(plain_input): - filtered_input_items.append(run_item) + forward_verbatim = _should_forward_new_item(plain_input) + normalized_new_items.append((plain_input, forward_verbatim, run_item)) + + normalized_items = normalized_pre_items + normalized_new_items + + owned_items: list[NestedHistoryOwnedItem] = [] + if history_mapper is not None: + transcript = flattened_history + [item for item, _, _ in normalized_items] + history_items = history_mapper(transcript) + else: + history_items, owned_items = _build_ordered_default_history( + flattened_history, + normalized_items, + ) + + copied_history = [deepcopy(item) for item in history_items] + owned_items = [ + replace( + owned_item, + input_item=copied_history[owned_item.input_index], + ) + for owned_item in owned_items + ] + + nested = handoff_input_data.clone( + input_history=tuple(copied_history), + pre_handoff_items=(), + # The mapped history is the exact model input. New items stay unchanged for session + # history. + input_items=(), + ) + object.__setattr__(nested, "_nested_history_owned_items", tuple(owned_items)) - transcript = flattened_history + pre_items_as_inputs + new_items_as_inputs + return nested, tuple(owned_items) - mapper = history_mapper or default_handoff_history_mapper - history_items = mapper(transcript) - return handoff_input_data.clone( - input_history=tuple(deepcopy(item) for item in history_items), - pre_handoff_items=tuple(filtered_pre_items), - # new_items stays unchanged for session history. - input_items=tuple(filtered_input_items), +def _get_nested_history_owned_items( + handoff_input_data: HandoffInputData, + *, + source_data: HandoffInputData | None = None, +) -> tuple[NestedHistoryOwnedItem, ...]: + """Match clean nested input occurrences to their source run items.""" + from ..run_internal.items import ( + NestedHistoryOwnedItem, + digest_input_item, ) + if isinstance(handoff_input_data.input_history, str): + return () + + declared_items = tuple( + item + for item in getattr(handoff_input_data, "_nested_history_owned_items", ()) + if isinstance(item, NestedHistoryOwnedItem) + ) + if not declared_items: + return () + + current_by_source_id: dict[int, RunItem] = {} + if source_data is not None: + current_items = ( + *handoff_input_data.pre_handoff_items, + *handoff_input_data.new_items, + ) + source_items = (*source_data.pre_handoff_items, *source_data.new_items) + mapped_items = _map_run_item_occurrences(current_items, source_items) + current_by_source_id = { + id(source_item): current_item + for current_item, source_item in zip(current_items, mapped_items, strict=True) + if source_item is not None + } + + input_digests = [digest_input_item(item) for item in handoff_input_data.input_history] + input_digest_counts: dict[str, int] = {} + for digest in input_digests: + if digest is not None: + input_digest_counts[digest] = input_digest_counts.get(digest, 0) + 1 + owned_digest_counts: dict[str, int] = {} + for owned_item in declared_items: + owned_digest_counts[owned_item.digest] = owned_digest_counts.get(owned_item.digest, 0) + 1 + + retained: list[NestedHistoryOwnedItem] = [] + used_input_indexes: set[int] = set() + for owned_item in declared_items: + input_index = next( + ( + index + for index, item in enumerate(handoff_input_data.input_history) + if index not in used_input_indexes + and owned_item.input_item is not None + and item is owned_item.input_item + and input_digests[index] == owned_item.digest + ), + None, + ) + if input_index is None: + candidate_indexes = [ + index + for index, digest in enumerate(input_digests) + if index not in used_input_indexes and digest == owned_item.digest + ] + all_equal_occurrences_owned = ( + input_digest_counts.get(owned_item.digest, 0) + == owned_digest_counts[owned_item.digest] + ) + if len(candidate_indexes) == 1 or all_equal_occurrences_owned: + input_index = ( + owned_item.input_index + if owned_item.input_index in candidate_indexes + else candidate_indexes[0] + if candidate_indexes + else None + ) + if input_index is None: + continue + used_input_indexes.add(input_index) + input_item = handoff_input_data.input_history[input_index] + source_run_item = ( + current_by_source_id.get(id(owned_item.run_item), owned_item.run_item) + if owned_item.run_item is not None + else None + ) + retained.append( + replace( + owned_item, + run_item=source_run_item, + input_index=input_index, + input_item=input_item, + ) + ) + return tuple(retained) + + +def _map_run_item_occurrences( + current_items: tuple[RunItem, ...], + source_items: tuple[RunItem, ...], +) -> list[RunItem | None]: + """Map copied filtered items back to original handoff occurrences when possible.""" + if not source_items: + return [None] * len(current_items) + + from ..run_internal.items import nested_history_run_item_occurrence_key + + used_source_indexes: set[int] = set() + mapped: list[RunItem | None] = [] + for current_item in current_items: + source_index = next( + ( + index + for index, source_item in enumerate(source_items) + if index not in used_source_indexes and source_item is current_item + ), + None, + ) + current_key = nested_history_run_item_occurrence_key(current_item) + if source_index is None and current_key is not None: + candidate_indexes = [ + index + for index, source_item in enumerate(source_items) + if index not in used_source_indexes + and nested_history_run_item_occurrence_key(source_item) == current_key + ] + if candidate_indexes: + source_index = candidate_indexes[0] + if source_index is None: + mapped.append(None) + continue + used_source_indexes.add(source_index) + mapped.append(source_items[source_index]) + return mapped + def default_handoff_history_mapper( transcript: list[TResponseInputItem], @@ -138,7 +309,51 @@ def _normalize_input_history( def _run_item_to_plain_input(run_item: RunItem) -> TResponseInputItem: - return deepcopy(run_item.to_input_item()) + from ..run_internal.items import run_item_to_input_item + + input_item = run_item_to_input_item(run_item) + if input_item is None: + raise TypeError(f"Unsupported nested handoff run item: {run_item.type}") + return deepcopy(input_item) + + +def _build_ordered_default_history( + flattened_history: list[TResponseInputItem], + normalized_items: list[tuple[TResponseInputItem, bool, RunItem]], +) -> tuple[list[TResponseInputItem], list[NestedHistoryOwnedItem]]: + from ..run_internal.items import ( + NestedHistoryOwnedItem, + digest_input_item, + ensure_nested_history_run_item_occurrence_key, + ) + + history_items: list[TResponseInputItem] = [] + owned_items: list[NestedHistoryOwnedItem] = [] + pending_summary = list(flattened_history) + + for plain_input, forward_verbatim, run_item in normalized_items: + if not forward_verbatim: + pending_summary.append(plain_input) + continue + if pending_summary or not history_items: + history_items.extend(default_handoff_history_mapper(pending_summary)) + pending_summary = [] + digest = digest_input_item(plain_input) + if digest is not None: + ensure_nested_history_run_item_occurrence_key(run_item) + owned_items.append( + NestedHistoryOwnedItem( + run_item=run_item, + input_index=len(history_items), + digest=digest, + ) + ) + history_items.append(plain_input) + + if pending_summary or not history_items: + history_items.extend(default_handoff_history_mapper(pending_summary)) + + return history_items, owned_items def _build_summary_message(transcript: list[TResponseInputItem]) -> TResponseInputItem: @@ -167,6 +382,7 @@ def _build_summary_message(transcript: list[TResponseInputItem]) -> TResponseInp def _format_transcript_item(item: TResponseInputItem) -> str: + item = _strip_transcript_item_metadata(item) role = item.get("role") if isinstance(role, str): content = item.get("content") @@ -334,7 +550,7 @@ def _parse_summary_json_item(value: str) -> TResponseInputItem | None: if not isinstance(parsed, dict): return None parsed.pop("provider_data", None) - return cast(TResponseInputItem, parsed) + return _strip_transcript_item_metadata(cast(TResponseInputItem, parsed)) def _parse_legacy_typed_item(item_type: str, content: str) -> TResponseInputItem | None: @@ -348,7 +564,14 @@ def _parse_legacy_typed_item(item_type: str, content: str) -> TResponseInputItem return None parsed.pop("provider_data", None) parsed["type"] = item_type - return cast(TResponseInputItem, parsed) + return _strip_transcript_item_metadata(cast(TResponseInputItem, parsed)) + + +def _strip_transcript_item_metadata(item: TResponseInputItem) -> TResponseInputItem: + """Remove SDK-only fields before nested transcript formatting or replay.""" + from ..run_internal.items import strip_internal_input_item_metadata + + return strip_internal_input_item_metadata(item) def _split_role_and_name(role_text: str) -> tuple[str, str | None]: diff --git a/src/agents/result.py b/src/agents/result.py index 8ae407003a..f63a8e7f68 100644 --- a/src/agents/result.py +++ b/src/agents/result.py @@ -5,7 +5,7 @@ import copy import weakref from collections.abc import AsyncIterator -from dataclasses import InitVar, dataclass, field +from dataclasses import InitVar, dataclass, field, replace from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast from pydantic import GetCoreSchemaHandler @@ -30,7 +30,14 @@ ) from .logger import logger from .run_context import RunContextWrapper -from .run_internal.items import run_items_to_input_items +from .run_internal.items import ( + NestedHistoryOwnedItemRef, + digest_input_item, + filter_nested_history_owned_item_refs_for_input, + rebase_nested_history_owned_item_refs, + resolve_nested_history_owned_item_indexes, + run_items_to_input_items, +) from .run_internal.run_steps import ( NextStepInterruption, ProcessedResponse, @@ -68,6 +75,33 @@ class AgentToolInvocation: """The raw JSON arguments for the nested invocation.""" +def _reconciled_result_owned_item_refs( + result: RunResultBase, + public_input: str | list[TResponseInputItem], +) -> list[NestedHistoryOwnedItemRef]: + """Retain ownership only for the exact public-input occurrences.""" + owned_item_refs = getattr(result, "_nested_history_owned_session_item_refs", []) + return filter_nested_history_owned_item_refs_for_input( + public_input, + owned_item_refs, + ) + + +def _state_snapshot_owned_item_refs( + result: RunResultBase, + state_input: str | list[TResponseInputItem], +) -> list[NestedHistoryOwnedItemRef]: + """Rebind validated ownership coordinates to the input snapshot stored in RunState.""" + if isinstance(state_input, str): + return [] + return [ + replace(item_ref, input_item=state_input[item_ref.input_index]) + for item_ref in getattr(result, "_nested_history_owned_session_item_refs", []) + if 0 <= item_ref.input_index < len(state_input) + and digest_input_item(state_input[item_ref.input_index]) == item_ref.digest + ] + + def _populate_state_from_result( state: RunState[Any], result: RunResultBase, @@ -88,6 +122,13 @@ def _populate_state_from_result( else: state._generated_items = result.new_items state._session_items = list(result.new_items) + snapshot_refs = _state_snapshot_owned_item_refs(result, state._original_input) + live_refs = rebase_nested_history_owned_item_refs( + state._original_input, + state._session_items, + snapshot_refs, + ) + state._nested_history_owned_session_item_refs = live_refs state._model_responses = result.raw_responses state._input_guardrail_results = result.input_guardrail_results state._output_guardrail_results = result.output_guardrail_results @@ -127,11 +168,35 @@ def _populate_state_from_result( ToInputListMode = Literal["preserve_all", "normalized"] +def _preserve_all_session_items( + result: RunResultBase, + reasoning_item_id_policy: Literal["preserve", "omit"] | None, + public_input: str | list[TResponseInputItem], + owned_item_refs: list[NestedHistoryOwnedItemRef], +) -> list[TResponseInputItem]: + """Avoid replaying session items already moved into ordered nested history.""" + retained_refs = filter_nested_history_owned_item_refs_for_input( + public_input, + owned_item_refs, + ) + excluded = resolve_nested_history_owned_item_indexes( + result.new_items, + retained_refs, + ) + if not excluded: + return run_items_to_input_items(result.new_items, reasoning_item_id_policy) + + filtered_items = [item for index, item in enumerate(result.new_items) if index not in excluded] + return run_items_to_input_items(filtered_items, reasoning_item_id_policy) + + def _input_items_for_result( result: RunResultBase, *, mode: ToInputListMode, reasoning_item_id_policy: Literal["preserve", "omit"] | None, + public_input: str | list[TResponseInputItem], + owned_item_refs: list[NestedHistoryOwnedItemRef], ) -> list[TResponseInputItem]: """Return input items for the requested result view. @@ -139,11 +204,16 @@ def _input_items_for_result( the canonical continuation input when handoff filtering rewrote model history, otherwise it falls back to the same converted history. """ - session_items = run_items_to_input_items(result.new_items, reasoning_item_id_policy) if mode == "preserve_all": - return session_items + return _preserve_all_session_items( + result, + reasoning_item_id_policy, + public_input, + owned_item_refs, + ) if mode != "normalized": raise ValueError(f"Unsupported to_input_list mode: {mode}") + session_items = run_items_to_input_items(result.new_items, reasoning_item_id_policy) if not getattr(result, "_replay_from_model_input_items", False): # Most runs never rewrite continuation history, so normalized stays identical to the # historical preserve-all view unless the runner explicitly marked a divergence. @@ -213,6 +283,12 @@ class RunResultBase(abc.ABC): This is only set when the runner preserved extra session history items that should not be replayed into the next local run, such as nested handoff history or filtered handoff input. """ + _nested_history_owned_session_item_refs: list[NestedHistoryOwnedItemRef] = field( + default_factory=list, + init=False, + repr=False, + ) + """Session item occurrences already represented verbatim in SDK-default nested history.""" _sandbox_resume_state: dict[str, object] | None = field(default=None, init=False, repr=False) """Serialized sandbox session state captured during the run.""" _sandbox_session: BaseSandboxSession | None = field(default=None, init=False, repr=False) @@ -295,12 +371,16 @@ def to_input_list( full plain-item history. ``mode="normalized"`` prefers the canonical continuation input when handoff filtering rewrote model history, while remaining identical for ordinary runs. """ - original_items: list[TResponseInputItem] = ItemHelpers.input_to_new_input_list(self.input) + public_input = self.input + owned_item_refs = _reconciled_result_owned_item_refs(self, public_input) + original_items = ItemHelpers.input_to_new_input_list(public_input) reasoning_item_id_policy = getattr(self, "_reasoning_item_id_policy", None) replay_items = _input_items_for_result( self, mode=mode, reasoning_item_id_policy=reasoning_item_id_policy, + public_input=public_input, + owned_item_refs=owned_item_refs, ) return original_items + replay_items diff --git a/src/agents/run.py b/src/agents/run.py index f28462d461..cbee4012b2 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -77,6 +77,7 @@ from .run_internal.items import ( copy_input_items, normalize_resumed_input, + reconcile_nested_history_owned_input_after_rewrite, ) from .run_internal.oai_conversation import OpenAIServerConversationTracker from .run_internal.prompt_cache_key import PromptCacheKeyResolver @@ -104,6 +105,7 @@ from .run_internal.session_persistence import ( persist_session_items_for_guardrail_trip, prepare_input_with_session, + reconcile_nested_history_owned_session_item_refs, resumed_turn_items, save_result_to_session, save_resumed_turn_items, @@ -618,6 +620,15 @@ async def run( current_turn = run_state._current_turn raw_original_input = run_state._original_input original_input = normalize_resumed_input(raw_original_input) + ( + original_input, + run_state._nested_history_owned_session_item_refs, + ) = reconcile_nested_history_owned_input_after_rewrite( + raw_original_input, + original_input, + run_state._nested_history_owned_session_item_refs, + ) + run_state._original_input = copy_input_items(original_input) generated_items = run_state._generated_items session_items = list(run_state._session_items) model_responses = run_state._model_responses @@ -697,6 +708,9 @@ def _finalize_result(result: RunResult) -> RunResult: finalized_result._generated_prompt_cache_key = ( run_state._generated_prompt_cache_key ) + finalized_result._nested_history_owned_session_item_refs = list( + run_state._nested_history_owned_session_item_refs + ) completed_result = finalized_result return finalized_result @@ -804,6 +818,7 @@ def _finalize_result(result: RunResult) -> RunResult: current_bindings = bind_public_agent(current_agent) execution_agent = current_bindings.execution_agent + input_before_sandbox = copy_input_items(original_input) prepared_sandbox = await sandbox_runtime.prepare_agent( current_agent=current_agent, current_input=original_input, @@ -812,9 +827,19 @@ def _finalize_result(result: RunResult) -> RunResult: ) current_bindings = prepared_sandbox.bindings execution_agent = current_bindings.execution_agent - original_input = copy_input_items(prepared_sandbox.input) + if run_state is not None: + ( + original_input, + run_state._nested_history_owned_session_item_refs, + ) = reconcile_nested_history_owned_input_after_rewrite( + input_before_sandbox, + prepared_sandbox.input, + run_state._nested_history_owned_session_item_refs, + ) + else: + original_input = copy_input_items(prepared_sandbox.input) if starting_input is not None and not isinstance(starting_input, RunState): - starting_input = copy_input_items(prepared_sandbox.input) + starting_input = copy_input_items(original_input) if run_state is not None: run_state._original_input = copy_input_items(original_input) @@ -866,10 +891,21 @@ def _finalize_result(result: RunResult) -> RunResult: run_state._last_processed_response, ) + input_before_turn_rewrite = original_input original_input = turn_result.original_input generated_items, turn_session_items = resumed_turn_items(turn_result) session_items.extend(turn_session_items) if run_state is not None: + if turn_result.nested_history_owned_items is not None: + run_state._nested_history_owned_session_item_refs = ( + reconcile_nested_history_owned_session_item_refs( + session_items, + run_state._nested_history_owned_session_item_refs, + input_before_turn_rewrite, + turn_result.original_input, + turn_result.nested_history_owned_items, + ) + ) update_run_state_after_resume( run_state, turn_result=turn_result, @@ -1284,12 +1320,23 @@ def _finalize_result(result: RunResult) -> RunResult: should_run_agent_start_hooks = False model_responses.append(turn_result.model_response) + input_before_turn_rewrite = original_input original_input = turn_result.original_input # For model input, use new_step_items (filtered on handoffs). generated_items = turn_result.pre_step_items + turn_result.new_step_items # Accumulate unfiltered items for observability. turn_session_items = session_items_for_turn(turn_result) session_items.extend(turn_session_items) + if run_state is not None and turn_result.nested_history_owned_items is not None: + run_state._nested_history_owned_session_item_refs = ( + reconcile_nested_history_owned_session_item_refs( + session_items, + run_state._nested_history_owned_session_item_refs, + input_before_turn_rewrite, + turn_result.original_input, + turn_result.nested_history_owned_items, + ) + ) if server_conversation_tracker is not None: pending_server_items = list(turn_result.new_step_items) server_conversation_tracker.track_server_items(turn_result.model_response) @@ -1705,6 +1752,15 @@ def run_streamed( # primeFromState will mark items as sent so prepareInput skips them raw_input_for_result = run_state._original_input input_for_result = normalize_resumed_input(raw_input_for_result) + ( + input_for_result, + run_state._nested_history_owned_session_item_refs, + ) = reconcile_nested_history_owned_input_after_rewrite( + raw_input_for_result, + input_for_result, + run_state._nested_history_owned_session_item_refs, + ) + run_state._original_input = copy_input_items(input_for_result) # Use context from RunState if not provided, otherwise override it. context_wrapper = resolve_resumed_context( run_state=run_state, diff --git a/src/agents/run_config.py b/src/agents/run_config.py index 45dcca5b10..08ee4cff9e 100644 --- a/src/agents/run_config.py +++ b/src/agents/run_config.py @@ -234,9 +234,10 @@ class RunConfig: """ nest_handoff_history: bool = False - """Opt-in beta: wrap prior run history in a single assistant message before handing off when no - custom input filter is set. This is disabled by default while we stabilize nested handoffs; set - to True to enable the collapsed transcript behavior. Server-managed conversations + """Opt-in beta: compact prior run history into ordered assistant summary segments while + preserving lossless message items in their original positions. This is disabled by default + while we stabilize nested handoffs; set to True to enable the compacted transcript behavior. + Server-managed conversations (`conversation_id`, `previous_response_id`, or `auto_previous_response_id`) automatically disable this behavior with a warning. """ @@ -244,7 +245,8 @@ class RunConfig: handoff_history_mapper: HandoffHistoryMapper | None = None """Optional function that receives the normalized transcript (history + handoff items) and returns the input history that should be passed to the next agent. When left as `None`, the - runner collapses the transcript into a single assistant message. This function only runs when + runner uses ordered summary segments around lossless message items. When supplied, the + function's return value is used as the exact input history. This function only runs when `nest_handoff_history` is True. """ diff --git a/src/agents/run_internal/items.py b/src/agents/run_internal/items.py index aadba1d361..b8d0769a9c 100644 --- a/src/agents/run_internal/items.py +++ b/src/agents/run_internal/items.py @@ -5,9 +5,12 @@ from __future__ import annotations +import hashlib import json from collections.abc import Sequence +from dataclasses import dataclass, field, replace from typing import Any, Literal, cast +from uuid import uuid4 from openai.types.responses import ResponseFunctionToolCall from pydantic import BaseModel @@ -20,6 +23,7 @@ REJECTION_MESSAGE = DEFAULT_APPROVAL_REJECTION_MESSAGE TOOL_CALL_SESSION_DESCRIPTION_KEY = "_agents_tool_description" TOOL_CALL_SESSION_TITLE_KEY = "_agents_tool_title" +_NESTED_HISTORY_RUN_ITEM_OCCURRENCE_KEY = "_agents_nested_history_occurrence_key" _TOOL_CALL_TO_OUTPUT_TYPE: dict[str, str] = { "function_call": "function_call_output", "custom_tool_call": "custom_tool_call_output", @@ -31,6 +35,8 @@ } __all__ = [ + "NestedHistoryOwnedItemRef", + "NestedHistoryOwnedItem", "ReasoningItemIdPolicy", "REJECTION_MESSAGE", "TOOL_CALL_SESSION_DESCRIPTION_KEY", @@ -44,6 +50,13 @@ "normalize_input_items_for_api", "normalize_resumed_input", "fingerprint_input_item", + "digest_input_item", + "ensure_nested_history_run_item_occurrence_key", + "nested_history_run_item_occurrence_key", + "reconcile_nested_history_owned_input_after_rewrite", + "filter_nested_history_owned_item_refs_for_input", + "rebase_nested_history_owned_item_refs", + "resolve_nested_history_owned_item_indexes", "deduplicate_input_items", "deduplicate_input_items_preferring_latest", "strip_internal_input_item_metadata", @@ -55,6 +68,44 @@ ] +@dataclass(frozen=True) +class NestedHistoryOwnedItem: + """A run item and the exact nested-input occurrence that represents it.""" + + run_item: RunItem | None + input_index: int + digest: str + input_item: TResponseInputItem | None = field(default=None, compare=False, repr=False) + + +@dataclass(frozen=True) +class NestedHistoryOwnedItemRef: + """Durable coordinates plus the live object for one owned session occurrence.""" + + session_index: int + digest: str + input_index: int + run_item: RunItem | None = field(default=None, compare=False, repr=False) + input_item: TResponseInputItem | None = field(default=None, compare=False, repr=False) + + +def nested_history_run_item_occurrence_key(run_item: RunItem | None) -> str | None: + """Return the private copy-lineage key for a run item, when one exists.""" + if run_item is None: + return None + key = getattr(run_item, _NESTED_HISTORY_RUN_ITEM_OCCURRENCE_KEY, None) + return key if isinstance(key, str) and key else None + + +def ensure_nested_history_run_item_occurrence_key(run_item: RunItem) -> str: + """Bind an ephemeral key that survives object copies but never enters model payloads.""" + key = nested_history_run_item_occurrence_key(run_item) + if key is None: + key = uuid4().hex + setattr(run_item, _NESTED_HISTORY_RUN_ITEM_OCCURRENCE_KEY, key) + return key + + ReasoningItemIdPolicy = Literal["preserve", "omit"] @@ -265,6 +316,223 @@ def fingerprint_input_item(item: Any, *, ignore_ids_for_matching: bool = False) return None +def digest_input_item(item: Any) -> str | None: + """Return a fixed-size digest of an input item for durable occurrence tracking.""" + coerced = _coerce_to_dict(item) + if coerced is not None: + coerced = cast( + dict[str, Any], + strip_internal_input_item_metadata(cast(TResponseInputItem, coerced)), + ) + if coerced.get("role") == "assistant" and coerced.get("status") in {None, "completed"}: + coerced.pop("status", None) + item = coerced + + fingerprint = fingerprint_input_item(item) + if fingerprint is None: + return None + return hashlib.sha256(fingerprint.encode("utf-8")).hexdigest() + + +def filter_nested_history_owned_item_refs_for_input( + input: str | Sequence[TResponseInputItem], + owned_item_refs: Sequence[NestedHistoryOwnedItemRef], +) -> list[NestedHistoryOwnedItemRef]: + """Keep ownership whose exact clean input occurrence is still present.""" + if isinstance(input, str) or not owned_item_refs: + return [] + + retained: list[NestedHistoryOwnedItemRef] = [] + used_input_indexes: set[int] = set() + for item_ref in owned_item_refs: + input_index = next( + ( + index + for index, item in enumerate(input) + if index not in used_input_indexes + and item_ref.input_item is not None + and item is item_ref.input_item + and digest_input_item(item) == item_ref.digest + ), + None, + ) + if input_index is None and item_ref.input_item is None: + candidate_index = item_ref.input_index + if ( + 0 <= candidate_index < len(input) + and candidate_index not in used_input_indexes + and digest_input_item(input[candidate_index]) == item_ref.digest + ): + input_index = candidate_index + if input_index is None: + continue + used_input_indexes.add(input_index) + retained.append(replace(item_ref, input_index=input_index, input_item=input[input_index])) + return retained + + +def reconcile_nested_history_owned_input_after_rewrite( + previous_input: str | Sequence[TResponseInputItem], + rewritten_input: str | Sequence[TResponseInputItem], + owned_item_refs: Sequence[NestedHistoryOwnedItemRef], +) -> tuple[str | list[TResponseInputItem], list[NestedHistoryOwnedItemRef]]: + """Rebind ownership after an unambiguous input rewrite.""" + if isinstance(rewritten_input, str) or not owned_item_refs: + return ( + rewritten_input if isinstance(rewritten_input, str) else list(rewritten_input), + [], + ) + if isinstance(previous_input, str): + return list(rewritten_input), [] + + rewritten = list(rewritten_input) + previous = list(previous_input) + previous_digests = [digest_input_item(item) for item in previous] + rewritten_digests = [digest_input_item(item) for item in rewritten] + recoverable_ref_counts: dict[str, int] = {} + for item_ref in owned_item_refs: + if item_ref.input_item is not None and any( + item is item_ref.input_item and previous_digests[index] == item_ref.digest + for index, item in enumerate(previous) + ): + recoverable_ref_counts[item_ref.digest] = ( + recoverable_ref_counts.get(item_ref.digest, 0) + 1 + ) + used_indexes: set[int] = set() + retained: list[NestedHistoryOwnedItemRef] = [] + + for item_ref in owned_item_refs: + identity_match = next( + ( + index + for index, item in enumerate(rewritten) + if index not in used_indexes + and item_ref.input_item is not None + and item is item_ref.input_item + and rewritten_digests[index] == item_ref.digest + ), + None, + ) + if identity_match is not None: + used_indexes.add(identity_match) + retained.append( + replace( + item_ref, + input_index=identity_match, + input_item=rewritten[identity_match], + ) + ) + continue + + previous_match = next( + ( + index + for index, item in enumerate(previous) + if item is item_ref.input_item and previous_digests[index] == item_ref.digest + ), + None, + ) + candidate_indexes = [ + index + for index, digest in enumerate(rewritten_digests) + if index not in used_indexes and digest == item_ref.digest + ] + previous_count = previous_digests.count(item_ref.digest) + rewritten_count = rewritten_digests.count(item_ref.digest) + all_equal_occurrences_owned = ( + previous_count == rewritten_count == recoverable_ref_counts.get(item_ref.digest, 0) + ) + if ( + previous_match is None + or not candidate_indexes + or not ((previous_count == 1 and rewritten_count == 1) or all_equal_occurrences_owned) + ): + continue + + candidate_index = candidate_indexes[0] + used_indexes.add(candidate_index) + retained.append( + replace( + item_ref, + input_index=candidate_index, + input_item=rewritten[candidate_index], + ) + ) + + return rewritten, retained + + +def resolve_nested_history_owned_item_indexes( + run_items: Sequence[RunItem], + owned_item_refs: Sequence[NestedHistoryOwnedItemRef], +) -> set[int]: + """Resolve ownership references without dropping a different item after list mutation.""" + if not owned_item_refs: + return set() + + resolved: set[int] = set() + for item_ref in owned_item_refs: + occurrence_key = nested_history_run_item_occurrence_key(item_ref.run_item) + candidate_indexes: list[int] = [] + if 0 <= item_ref.session_index < len(run_items): + candidate_indexes.append(item_ref.session_index) + candidate_indexes.extend( + index + for index, item in enumerate(run_items) + if index != item_ref.session_index + and item_ref.run_item is not None + and ( + item is item_ref.run_item + or ( + occurrence_key is not None + and nested_history_run_item_occurrence_key(item) == occurrence_key + ) + ) + ) + for index in candidate_indexes: + if index in resolved or item_ref.run_item is None: + continue + if run_items[index] is not item_ref.run_item and not ( + occurrence_key is not None + and nested_history_run_item_occurrence_key(run_items[index]) == occurrence_key + ): + continue + input_item = run_item_to_input_item(run_items[index]) + if input_item is not None and digest_input_item(input_item) == item_ref.digest: + resolved.add(index) + break + + return resolved + + +def rebase_nested_history_owned_item_refs( + input: str | Sequence[TResponseInputItem], + run_items: Sequence[RunItem], + owned_item_refs: Sequence[NestedHistoryOwnedItemRef], +) -> list[NestedHistoryOwnedItemRef]: + """Rebase surviving ownership onto exact live input and session occurrences.""" + retained_refs = filter_nested_history_owned_item_refs_for_input(input, owned_item_refs) + rebased: list[NestedHistoryOwnedItemRef] = [] + used_indexes: set[int] = set() + for item_ref in retained_refs: + occurrence_key = nested_history_run_item_occurrence_key(item_ref.run_item) + for index, run_item in enumerate(run_items): + if index in used_indexes or item_ref.run_item is None: + continue + if run_item is not item_ref.run_item and not ( + occurrence_key is not None + and nested_history_run_item_occurrence_key(run_item) == occurrence_key + ): + continue + input_item = run_item_to_input_item(run_item) + if input_item is None or digest_input_item(input_item) != item_ref.digest: + continue + used_indexes.add(index) + rebased.append(replace(item_ref, session_index=index, run_item=run_item)) + break + return rebased + + def _dedupe_key(item: TResponseInputItem) -> str | None: """Return a stable identity key when items carry explicit identifiers.""" payload = _coerce_to_dict(item) diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index b85347f6bf..7d33dba525 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -118,6 +118,7 @@ ensure_input_item_format, normalize_resumed_input, prepare_model_input_items, + reconcile_nested_history_owned_input_after_rewrite, run_items_to_input_items, ) from .model_retry import ( @@ -146,6 +147,7 @@ from .session_persistence import ( persist_session_items_for_guardrail_trip, prepare_input_with_session, + reconcile_nested_history_owned_session_item_refs, resumed_turn_items, rewind_session_items, save_result_to_session, @@ -548,6 +550,9 @@ def _sync_conversation_tracking_from_tracker() -> None: streamed_result._state = run_state if run_state is not None: streamed_result._model_input_items = list(run_state._generated_items) + streamed_result._nested_history_owned_session_item_refs = list( + run_state._nested_history_owned_session_item_refs + ) # Streamed follow-ups need the same normalized replay signal as sync runs when the # runner's continuation differs from the richer session history. streamed_result._replay_from_model_input_items = list( @@ -602,6 +607,18 @@ def _sync_conversation_tracking_from_tracker() -> None: prepared_input: str | list[TResponseInputItem] if is_resumed_state and run_state is not None: prepared_input = normalize_resumed_input(starting_input) + ( + prepared_input, + run_state._nested_history_owned_session_item_refs, + ) = reconcile_nested_history_owned_input_after_rewrite( + starting_input, + prepared_input, + run_state._nested_history_owned_session_item_refs, + ) + run_state._original_input = copy_input_items(prepared_input) + streamed_result._nested_history_owned_session_item_refs = list( + run_state._nested_history_owned_session_item_refs + ) streamed_result.input = prepared_input streamed_result._original_input_for_persistence = [] streamed_result._stream_input_persisted = True @@ -723,6 +740,7 @@ async def _save_stream_items_without_count( sequential_guardrails = [] if sandbox_runtime is not None: + input_before_sandbox = copy_input_items(prepared_turn_input) prepared_sandbox = await sandbox_runtime.prepare_agent( current_agent=current_agent, current_input=prepared_turn_input, @@ -731,11 +749,19 @@ async def _save_stream_items_without_count( ) current_bindings = prepared_sandbox.bindings execution_agent = current_bindings.execution_agent - prepared_turn_input = copy_input_items(prepared_sandbox.input) + prepared_turn_input, retained_owned_refs = ( + reconcile_nested_history_owned_input_after_rewrite( + input_before_sandbox, + prepared_sandbox.input, + streamed_result._nested_history_owned_session_item_refs, + ) + ) + streamed_result._nested_history_owned_session_item_refs = retained_owned_refs streamed_result.input = prepared_turn_input streamed_result._original_input = copy_input_items(prepared_turn_input) if run_state is not None: run_state._original_input = copy_input_items(prepared_turn_input) + run_state._nested_history_owned_session_item_refs = list(retained_owned_refs) sandbox_runtime.apply_result_metadata(streamed_result) if is_resumed_state and run_state is not None and run_state._current_step is not None: @@ -770,6 +796,7 @@ async def _save_stream_items_without_count( ), ) + input_before_turn_rewrite = streamed_result.input streamed_result.input = turn_result.original_input streamed_result._original_input = copy_input_items(turn_result.original_input) generated_items, turn_session_items = resumed_turn_items(turn_result) @@ -778,6 +805,17 @@ async def _save_stream_items_without_count( ) streamed_result._model_input_items = generated_items streamed_result.new_items = base_session_items + list(turn_session_items) + if turn_result.nested_history_owned_items is not None: + owned_refs = reconcile_nested_history_owned_session_item_refs( + streamed_result.new_items, + streamed_result._nested_history_owned_session_item_refs, + input_before_turn_rewrite, + turn_result.original_input, + turn_result.nested_history_owned_items, + ) + streamed_result._nested_history_owned_session_item_refs = owned_refs + if run_state is not None: + run_state._nested_history_owned_session_item_refs = list(owned_refs) streamed_result._replay_from_model_input_items = list( streamed_result._model_input_items ) != list(streamed_result.new_items) @@ -1073,6 +1111,7 @@ async def _save_stream_items_without_count( streamed_result.raw_responses = streamed_result.raw_responses + [ turn_result.model_response ] + input_before_turn_rewrite = streamed_result.input streamed_result.input = turn_result.original_input if isinstance(turn_result.next_step, NextStepHandoff): streamed_result._original_input = copy_input_items(turn_result.original_input) @@ -1083,6 +1122,17 @@ async def _save_stream_items_without_count( ) turn_session_items = session_items_for_turn(turn_result) streamed_result.new_items.extend(turn_session_items) + if turn_result.nested_history_owned_items is not None: + owned_refs = reconcile_nested_history_owned_session_item_refs( + streamed_result.new_items, + streamed_result._nested_history_owned_session_item_refs, + input_before_turn_rewrite, + turn_result.original_input, + turn_result.nested_history_owned_items, + ) + streamed_result._nested_history_owned_session_item_refs = owned_refs + if run_state is not None: + run_state._nested_history_owned_session_item_refs = list(owned_refs) streamed_result._replay_from_model_input_items = list( streamed_result._model_input_items ) != list(streamed_result.new_items) diff --git a/src/agents/run_internal/run_steps.py b/src/agents/run_internal/run_steps.py index ec64b760a6..98df09a416 100644 --- a/src/agents/run_internal/run_steps.py +++ b/src/agents/run_internal/run_steps.py @@ -26,6 +26,7 @@ ShellTool, ) from ..tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult +from .items import NestedHistoryOwnedItem __all__ = [ "QueueCompleteSentinel", @@ -202,6 +203,13 @@ class SingleStepResult: """Full unfiltered items for session history. When set, these are used instead of new_step_items for session saving and generated_items property.""" + nested_history_owned_items: list[NestedHistoryOwnedItem] | None = None + """Items moved verbatim into SDK-default nested history for this handoff. + + ``None`` means this step did not replace handoff history. A list means the handoff rewrote + history, so prior ownership must be reconciled against the new input before adding these items. + """ + output_guardrail_results: list[OutputGuardrailResult] = dataclasses.field(default_factory=list) """Output guardrail results (populated when a final output is produced).""" diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py index f483da13a3..f6a266d69b 100644 --- a/src/agents/run_internal/session_persistence.py +++ b/src/agents/run_internal/session_persistence.py @@ -25,13 +25,19 @@ from ..memory.openai_conversations_session import OpenAIConversationsSession from ..run_state import RunState from .items import ( + NestedHistoryOwnedItem, + NestedHistoryOwnedItemRef, ReasoningItemIdPolicy, copy_input_items, deduplicate_input_items_preferring_latest, + digest_input_item, drop_orphan_function_calls, ensure_input_item_format, + ensure_nested_history_run_item_occurrence_key, fingerprint_input_item, + nested_history_run_item_occurrence_key, normalize_input_items_for_api, + reconcile_nested_history_owned_input_after_rewrite, run_item_to_input_item, strip_internal_input_item_metadata, ) @@ -41,6 +47,8 @@ __all__ = [ "prepare_input_with_session", "persist_session_items_for_guardrail_trip", + "reconcile_nested_history_owned_session_item_refs", + "resolve_nested_history_owned_session_item_refs", "session_items_for_turn", "resumed_turn_items", "save_result_to_session", @@ -51,6 +59,83 @@ ] +def resolve_nested_history_owned_session_item_refs( + session_items: Sequence[RunItem], + current_input: str | Sequence[TResponseInputItem], + history_owned_items: Sequence[NestedHistoryOwnedItem], +) -> list[NestedHistoryOwnedItemRef]: + """Locate explicitly owned nested-history occurrences in full session history.""" + if not history_owned_items or isinstance(current_input, str): + return [] + + used_session_indexes: set[int] = set() + resolved: list[NestedHistoryOwnedItemRef] = [] + for owned_item in history_owned_items: + if owned_item.input_index >= len(current_input): + continue + input_item = current_input[owned_item.input_index] + if digest_input_item(input_item) != owned_item.digest: + continue + + occurrence_key = nested_history_run_item_occurrence_key(owned_item.run_item) + session_index = next( + ( + index + for index, session_item in enumerate(session_items) + if index not in used_session_indexes + and owned_item.run_item is not None + and ( + session_item is owned_item.run_item + or ( + occurrence_key is not None + and nested_history_run_item_occurrence_key(session_item) == occurrence_key + ) + ) + ), + None, + ) + if session_index is None: + continue + session_input = run_item_to_input_item(session_items[session_index]) + if session_input is None or digest_input_item(session_input) != owned_item.digest: + continue + used_session_indexes.add(session_index) + session_item = session_items[session_index] + ensure_nested_history_run_item_occurrence_key(session_item) + resolved.append( + NestedHistoryOwnedItemRef( + session_index=session_index, + digest=owned_item.digest, + input_index=owned_item.input_index, + run_item=session_item, + input_item=input_item, + ) + ) + return resolved + + +def reconcile_nested_history_owned_session_item_refs( + session_items: Sequence[RunItem], + previous_refs: Sequence[NestedHistoryOwnedItemRef], + previous_input: str | Sequence[TResponseInputItem], + current_input: str | Sequence[TResponseInputItem], + history_owned_items: Sequence[NestedHistoryOwnedItem], +) -> list[NestedHistoryOwnedItemRef]: + """Retain surviving ownership and add provenance introduced by a history rewrite.""" + _, retained_refs = reconcile_nested_history_owned_input_after_rewrite( + previous_input, + current_input, + previous_refs, + ) + new_refs = resolve_nested_history_owned_session_item_refs( + session_items, + current_input, + history_owned_items, + ) + retained_set = set(retained_refs) + return retained_refs + [item_ref for item_ref in new_refs if item_ref not in retained_set] + + async def prepare_input_with_session( input: str | list[TResponseInputItem], session: Session | None, diff --git a/src/agents/run_internal/turn_resolution.py b/src/agents/run_internal/turn_resolution.py index 33e1bee8e2..430ed8c9a3 100644 --- a/src/agents/run_internal/turn_resolution.py +++ b/src/agents/run_internal/turn_resolution.py @@ -44,6 +44,10 @@ from ..agent_tool_state import get_agent_tool_state_scope, peek_agent_tool_run_result from ..exceptions import ModelBehaviorError, ModelRefusalError, UserError from ..handoffs import Handoff, HandoffInputData, HandoffInputFilter, nest_handoff_history +from ..handoffs.history import ( + _get_nested_history_owned_items, + _nest_handoff_history_with_provenance, +) from ..items import ( CompactionItem, HandoffCallItem, @@ -99,6 +103,7 @@ ) from .items import ( REJECTION_MESSAGE, + NestedHistoryOwnedItem, apply_patch_rejection_item, function_rejection_item, shell_rejection_item, @@ -155,6 +160,8 @@ _select_function_tool_runs_for_resume, ) +_DEFAULT_NEST_HANDOFF_HISTORY = nest_handoff_history + __all__ = [ "execute_final_output_step", "execute_final_output", @@ -456,10 +463,22 @@ async def execute_handoffs( ) -> SingleStepResult: """Execute a handoff and prepare the next turn for the new agent.""" - def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffInputData: - if nest_handoff_history_fn is None: - return nest_handoff_history(data, history_mapper=mapper) - return nest_handoff_history_fn(data, mapper) + def nest_history( + data: HandoffInputData, + mapper: Any | None = None, + ) -> tuple[HandoffInputData, list[NestedHistoryOwnedItem]]: + if ( + nest_handoff_history_fn is None + and nest_handoff_history is _DEFAULT_NEST_HANDOFF_HISTORY + ): + nested, history_owned_items = _nest_handoff_history_with_provenance( + data, + history_mapper=mapper, + ) + return nested, list(history_owned_items) + if nest_handoff_history_fn is not None: + return nest_handoff_history_fn(data, mapper), [] + return nest_handoff_history(data, history_mapper=mapper), [] multiple_handoffs = len(run_handoffs) > 1 if multiple_handoffs: @@ -542,6 +561,7 @@ def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffIn ) handoff_input_data: HandoffInputData | None = None session_step_items: list[RunItem] | None = None + nested_history_owned_items: list[NestedHistoryOwnedItem] | None = None if input_filter or should_nest_history: handoff_input_data = HandoffInputData( input_history=tuple(original_input) @@ -598,8 +618,14 @@ def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffIn new_step_items = list(filtered.input_items) else: session_step_items = None + nested_history_owned_items = list( + _get_nested_history_owned_items(filtered, source_data=handoff_input_data) + ) elif should_nest_history and handoff_input_data is not None: - nested = nest_history(handoff_input_data, run_config.handoff_history_mapper) + nested, nested_history_owned_items = nest_history( + handoff_input_data, + run_config.handoff_history_mapper, + ) original_input = ( nested.input_history if isinstance(nested.input_history, str) @@ -626,6 +652,7 @@ def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffIn tool_input_guardrail_results=list(tool_input_guardrail_results or []), tool_output_guardrail_results=list(tool_output_guardrail_results or []), session_step_items=session_step_items, + nested_history_owned_items=nested_history_owned_items, ) @@ -947,11 +974,6 @@ async def resolve_interrupted_turn( execute_handoffs_call = execute_handoffs - def nest_history(data: HandoffInputData, mapper: Any | None = None) -> HandoffInputData: - if nest_handoff_history_fn is None: - return nest_handoff_history(data, history_mapper=mapper) - return nest_handoff_history_fn(data, mapper) - def _pending_approvals_from_state() -> list[ToolApprovalItem]: if ( run_state is not None @@ -1599,7 +1621,7 @@ def _add_unmatched_pending(approval: ToolApprovalItem) -> None: context_wrapper=context_wrapper, run_config=run_config, server_manages_conversation=server_manages_conversation, - nest_handoff_history_fn=nest_history, + nest_handoff_history_fn=nest_handoff_history_fn, tool_input_guardrail_results=tool_input_guardrail_results, tool_output_guardrail_results=tool_output_guardrail_results, ) diff --git a/src/agents/run_state.py b/src/agents/run_state.py index 5e5d7f0533..8362ec4439 100644 --- a/src/agents/run_state.py +++ b/src/agents/run_state.py @@ -80,6 +80,13 @@ ) from .logger import logger from .run_context import RunContextWrapper +from .run_internal.items import ( + NestedHistoryOwnedItemRef, + digest_input_item, + ensure_nested_history_run_item_occurrence_key, + nested_history_run_item_occurrence_key, + run_item_to_input_item, +) from .sandbox.capabilities.capability import Capability from .sandbox.session.base_sandbox_session import BaseSandboxSession from .tool import ( @@ -130,7 +137,7 @@ # 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.12" +CURRENT_SCHEMA_VERSION = "1.13" # 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.", @@ -149,6 +156,7 @@ "1.10": "Allows serialized RunState snapshots to disable max_turns with null.", "1.11": "Persists SDK-only custom data on tool output items across resume flows.", "1.12": "Persists input cache-write token usage across resume flows.", + "1.13": "Persists nested handoff history ownership across resume flows.", } SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES) @@ -224,6 +232,11 @@ class RunState(Generic[TContext, TAgent]): _session_items: list[RunItem] = field(default_factory=list) """Full, unfiltered run items for session history.""" + _nested_history_owned_session_item_refs: list[NestedHistoryOwnedItemRef] = field( + default_factory=list + ) + """Session-item occurrences also present verbatim in SDK-default nested input history.""" + _max_turns: int | None = 10 """Maximum allowed turns before forcing termination, or ``None`` for no limit.""" @@ -306,6 +319,7 @@ def __init__( self._model_responses = [] self._generated_items = [] self._session_items = [] + self._nested_history_owned_session_item_refs = [] self._input_guardrail_results = [] self._output_guardrail_results = [] self._tool_input_guardrail_results = [] @@ -413,6 +427,38 @@ def _serialize_original_input(self) -> str | list[Any]: normalized_items.append(normalized_item) return normalized_items + def _generated_session_item_indexes( + self, + generated_items: Sequence[RunItem], + ) -> list[int | None]: + """Map generated occurrences to the same live occurrences in session history.""" + used_session_indexes: set[int] = set() + indexes: list[int | None] = [] + for generated_item in generated_items: + session_index = next( + ( + index + for index, session_item in enumerate(self._session_items) + if index not in used_session_indexes and session_item is generated_item + ), + None, + ) + generated_key = nested_history_run_item_occurrence_key(generated_item) + if session_index is None and generated_key is not None: + session_index = next( + ( + index + for index, session_item in enumerate(self._session_items) + if index not in used_session_indexes + and nested_history_run_item_occurrence_key(session_item) == generated_key + ), + None, + ) + if session_index is not None: + used_session_indexes.add(session_index) + indexes.append(session_index) + return indexes + def _serialize_context_payload( self, *, @@ -713,6 +759,7 @@ def to_json( cast(Agent[Any], self._current_agent), agent_identity_keys_by_id=agent_identity_keys_by_id, ) + generated_items = self._merge_generated_items_with_processed() result = { "$schemaVersion": CURRENT_SCHEMA_VERSION, @@ -743,9 +790,17 @@ def to_json( "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, + "nested_history_owned_session_item_refs": [ + { + "index": item_ref.session_index, + "digest": item_ref.digest, + "input_index": item_ref.input_index, + } + for item_ref in self._nested_history_owned_session_item_refs + ], + "generated_session_item_indexes": self._generated_session_item_indexes(generated_items), } - 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 @@ -2071,16 +2126,23 @@ def _deserialize_tool_call_raw_item(normalized_raw_item: Mapping[str, Any]) -> A 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: + errors = exc.errors() + if not errors: return False - return missing_fields <= _ALLOWED_MISSING_MESSAGE_FIELDS + + for error in errors: + location = error.get("loc") + field = str(location[0]) if isinstance(location, tuple) and location else None + if error.get("type") == "missing" and field in _ALLOWED_MISSING_MESSAGE_FIELDS: + continue + if ( + error.get("type") == "literal_error" + and field == "status" + and error.get("input") is None + ): + continue + return False + return True def _deserialize_message_content_part(value: object) -> object: @@ -2524,8 +2586,9 @@ async def _build_run_state_from_json( 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", []), + serialized_generated_items = state_json.get("generated_items", []) + state._generated_items, generated_source_indexes = _deserialize_items_with_source_indexes( + serialized_generated_items, agent_map, agent_identity_map=agent_identity_map, ) @@ -2546,13 +2609,121 @@ async def _build_run_state_from_json( state._last_processed_response = None if "session_items" in state_json: - state._session_items = _deserialize_items( - state_json.get("session_items", []), + serialized_session_items = state_json.get("session_items", []) + state._session_items, session_source_indexes = _deserialize_items_with_source_indexes( + serialized_session_items, agent_map, agent_identity_map=agent_identity_map, ) else: + serialized_session_items = [] state._session_items = state._merge_generated_items_with_processed() + session_source_indexes = list(range(len(state._session_items))) + restored_session_indexes = { + source_index: restored_index + for restored_index, source_index in enumerate(session_source_indexes) + } + + generated_session_indexes = state_json.get("generated_session_item_indexes") + if generated_session_indexes is not None: + mapping_is_valid = not ( + not isinstance(serialized_generated_items, list) + or not isinstance(serialized_session_items, list) + or not isinstance(generated_session_indexes, list) + or len(generated_session_indexes) != len(serialized_generated_items) + ) + used_session_indexes: set[int] = set() + if mapping_is_valid: + for session_index_value in generated_session_indexes: + if session_index_value is None: + continue + if ( + type(session_index_value) is not int + or session_index_value < 0 + or session_index_value >= len(serialized_session_items) + or session_index_value in used_session_indexes + ): + mapping_is_valid = False + break + used_session_indexes.add(session_index_value) + + if not mapping_is_valid: + logger.warning("Ignoring invalid generated_session_item_indexes in serialized RunState") + else: + for restored_generated_index, generated_source_index in enumerate( + generated_source_indexes + ): + session_source_index = generated_session_indexes[generated_source_index] + if session_source_index is None: + continue + restored_session_index = restored_session_indexes.get(session_source_index) + if restored_session_index is None: + continue + if ( + serialized_generated_items[generated_source_index] + != serialized_session_items[session_source_index] + ): + logger.warning( + "Ignoring mismatched generated/session occurrence in serialized RunState" + ) + continue + state._generated_items[restored_generated_index] = state._session_items[ + restored_session_index + ] + + nested_history_refs_json = state_json.get("nested_history_owned_session_item_refs", []) + if not isinstance(nested_history_refs_json, list): + raise UserError( + "Run state nested_history_owned_session_item_refs must be a list of objects" + ) + nested_history_refs: list[NestedHistoryOwnedItemRef] = [] + for item_ref in nested_history_refs_json: + if ( + not isinstance(item_ref, Mapping) + or type(item_ref.get("index")) is not int + or cast(int, item_ref["index"]) < 0 + or not isinstance(item_ref.get("digest"), str) + or len(cast(str, item_ref["digest"])) != 64 + or type(item_ref.get("input_index")) is not int + or cast(int, item_ref["input_index"]) < 0 + ): + raise UserError( + "Run state nested_history_owned_session_item_refs entries must contain a " + "non-negative integer index and input_index, and 64-character digest" + ) + session_source_index = cast(int, item_ref["index"]) + input_index = cast(int, item_ref["input_index"]) + digest = cast(str, item_ref["digest"]) + if "session_items" in state_json and session_source_index >= len(serialized_session_items): + raise UserError("Run state nested history ownership references a missing session item") + session_index = restored_session_indexes.get(session_source_index) + if session_index is None: + logger.warning( + "Ignoring nested history ownership for skipped session item at index %s", + session_source_index, + ) + continue + if not isinstance(state._original_input, list) or input_index >= len(state._original_input): + raise UserError("Run state nested history ownership references a missing input item") + + run_item = state._session_items[session_index] + run_input_item = run_item_to_input_item(run_item) + if run_input_item is None or digest_input_item(run_input_item) != digest: + raise UserError("Run state nested history ownership session digest does not match") + ensure_nested_history_run_item_occurrence_key(run_item) + input_item = cast(TResponseInputItem, state._original_input[input_index]) + if digest_input_item(input_item) != digest: + raise UserError("Run state nested history ownership input digest does not match") + nested_history_refs.append( + NestedHistoryOwnedItemRef( + session_index=session_index, + digest=digest, + input_index=input_index, + run_item=run_item, + input_item=input_item, + ) + ) + state._nested_history_owned_session_item_refs = nested_history_refs state._mark_generated_items_merged_with_last_processed() @@ -3345,6 +3516,26 @@ def _resolve_agent_info( return result +def _deserialize_items_with_source_indexes( + items_data: list[dict[str, Any]], + agent_map: dict[str, Agent[Any]], + *, + agent_identity_map: Mapping[str, Agent[Any]] | None = None, +) -> tuple[list[RunItem], list[int]]: + """Deserialize items while retaining indexes of source entries that survived.""" + items: list[RunItem] = [] + source_indexes: list[int] = [] + for source_index, item_data in enumerate(items_data): + deserialized = _deserialize_items( + [item_data], + agent_map, + agent_identity_map=agent_identity_map, + ) + items.extend(deserialized) + source_indexes.extend([source_index] * len(deserialized)) + return items, source_indexes + + 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): diff --git a/tests/sandbox/test_runtime.py b/tests/sandbox/test_runtime.py index f455f26635..6c925766a6 100644 --- a/tests/sandbox/test_runtime.py +++ b/tests/sandbox/test_runtime.py @@ -25,6 +25,7 @@ from agents.items import ModelResponse, ToolCallOutputItem, TResponseInputItem from agents.model_settings import ModelSettings from agents.prompts import GenerateDynamicPromptData, Prompt +from agents.result import RunResult, RunResultStreaming from agents.run import CallModelData, ModelInputData, RunConfig from agents.run_context import AgentHookContext, RunContextWrapper from agents.run_state import RunState, _build_agent_identity_map @@ -787,6 +788,30 @@ def process_context(self, context: list[TResponseInputItem]) -> list[TResponseIn ] +class _DropContextTextCapability(Capability): + type: str = "drop-context-text" + text: str + + def __init__(self, text: str) -> None: + super().__init__(type="drop-context-text", **cast(Any, {"text": text})) + + def process_context(self, context: list[TResponseInputItem]) -> list[TResponseInputItem]: + return [item for item in context if self.text not in json.dumps(item, default=str)] + + +class _RebuildContextWithoutPrivateMetadataCapability(Capability): + type: str = "rebuild-context-without-private-metadata" + + def __init__(self) -> None: + super().__init__(type="rebuild-context-without-private-metadata") + + def process_context(self, context: list[TResponseInputItem]) -> list[TResponseInputItem]: + return [ + cast(TResponseInputItem, dict(item)) if isinstance(item, dict) else item + for item in context + ] + + class _SessionFileCapability(Capability): type: str = "session-files" bound_session: BaseSandboxSession | None = None @@ -1829,6 +1854,95 @@ async def test_runner_rebuilds_sandbox_resources_for_handoff_target_agent() -> N ) +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_context_rewrite_releases_removed_nested_history_ownership(streamed: bool) -> None: + triage_model = FakeModel() + worker_model = FakeModel(initial_output=[get_final_output_message("done")]) + client = _ManifestSessionClient() + worker = SandboxAgent( + name="worker", + model=worker_model, + default_manifest=Manifest(), + capabilities=[_DropContextTextCapability("handoff message")], + ) + triage = SandboxAgent( + name="triage", + model=triage_model, + default_manifest=Manifest(), + capabilities=[], + handoffs=[worker], + ) + triage_model.turn_outputs = [ + [get_final_output_message("handoff message"), get_handoff_tool_call(worker)] + ] + run_config = RunConfig( + sandbox=SandboxRunConfig(client=client), + nest_handoff_history=True, + ) + + result: RunResult | RunResultStreaming + if streamed: + result = Runner.run_streamed(triage, "route this", run_config=run_config) + async for _ in result.stream_events(): + pass + else: + result = await Runner.run(triage, "route this", run_config=run_config) + + replay_texts = [ + _extract_user_text(cast(dict[str, object], item)) + for item in result.to_input_list() + if isinstance(item, dict) and "content" in item + ] + assert replay_texts.count("handoff message") == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_context_rebuild_retains_unambiguous_nested_history_ownership( + streamed: bool, +) -> None: + triage_model = FakeModel() + worker_model = FakeModel(initial_output=[get_final_output_message("done")]) + client = _ManifestSessionClient() + worker = SandboxAgent( + name="worker", + model=worker_model, + default_manifest=Manifest(), + capabilities=[_RebuildContextWithoutPrivateMetadataCapability()], + ) + triage = SandboxAgent( + name="triage", + model=triage_model, + default_manifest=Manifest(), + capabilities=[], + handoffs=[worker], + ) + triage_model.turn_outputs = [ + [get_final_output_message("handoff message"), get_handoff_tool_call(worker)] + ] + run_config = RunConfig( + sandbox=SandboxRunConfig(client=client), + nest_handoff_history=True, + ) + + result: RunResult | RunResultStreaming + if streamed: + result = Runner.run_streamed(triage, "route this", run_config=run_config) + async for _ in result.stream_events(): + pass + else: + result = await Runner.run(triage, "route this", run_config=run_config) + + replay_texts = [ + _extract_user_text(cast(dict[str, object], item)) + for item in result.to_input_list() + if isinstance(item, dict) and "content" in item + ] + assert replay_texts.count("handoff message") == 1 + assert result._nested_history_owned_session_item_refs + + @pytest.mark.asyncio async def test_runner_resumed_handoff_materializes_manifest_for_new_sandbox_agent() -> None: triage_model = FakeModel() diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index 4b5ea867ce..0f3968917d 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -518,6 +518,21 @@ def _as_message(item: Any) -> dict[str, Any]: return cast(dict[str, Any], item) +def _input_message_text(item: Any) -> str: + message = _as_message(item) + content = message.get("content") + if isinstance(content, str): + return content + assert isinstance(content, list) + texts: list[str] = [] + for part in content: + assert isinstance(part, dict) + text = part.get("text") + if isinstance(text, str): + texts.append(text) + return "".join(texts) + + def _find_reasoning_input_item( items: str | list[TResponseInputItem] | Any, ) -> dict[str, Any] | None: @@ -1331,13 +1346,13 @@ async def test_structured_output(): assert result.final_output == Foo(bar="baz") assert len(result.raw_responses) == 4, "should have four model responses" - assert len(result.to_input_list()) == 10, ( - "should have input: conversation summary, function call, function call result, message, " - "handoff, handoff output, preamble message, tool call, tool call result, final output" + assert len(result.to_input_list()) == 11, ( + "should preserve ordered history segments plus function calls, messages, handoff items, " + "and the final output without replaying the carried-forward message twice" ) - assert len(result.to_input_list(mode="normalized")) == 6, ( + assert len(result.to_input_list(mode="normalized")) == 7, ( "should have normalized replay input: conversation summary, carried-forward message, " - "preamble message, tool call, tool call result, final output" + "handoff summary, preamble message, tool call, tool call result, final output" ) assert result.last_agent == agent_1, "should have handed off to agent_1" @@ -1414,14 +1429,21 @@ async def test_opt_in_handoff_history_nested_and_filters_respected(): ) assert isinstance(result.input, list) - assert len(result.input) == 1 + assert len(result.input) == 3 summary = _as_message(result.input[0]) assert summary["role"] == "assistant" summary_content = summary["content"] assert isinstance(summary_content, str) assert "" in summary_content - assert "triage summary" in summary_content + assert "triage summary" not in summary_content assert "user_message" in summary_content + assert _input_message_text(result.input[1]) == "triage summary" + handoff_summary = _input_message_text(result.input[2]) + assert "transfer_to_delegate" in handoff_summary + delegate_input = model.last_turn_args["input"] + assert isinstance(delegate_input, list) + assert len(delegate_input) == 3 + assert _input_message_text(delegate_input[1]) == "triage summary" passthrough_model = FakeModel() delegate = Agent(name="delegate", model=passthrough_model) @@ -1486,8 +1508,12 @@ async def test_opt_in_handoff_history_accumulates_across_multiple_handoffs(): assert isinstance(summary_content, str) assert summary_content.count("") == 1 assert "triage summary" in summary_content - assert "delegate update" in summary_content + assert "delegate update" not in summary_content assert "user_question" in summary_content + assert len(closer_input) == 3 + assert _input_message_text(closer_input[1]) == "delegate update" + handoff_summary = _input_message_text(closer_input[2]) + assert "transfer_to_closer" in handoff_summary @pytest.mark.asyncio diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index 8ee3a55db4..6cdb724d22 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -802,13 +802,13 @@ async def test_structured_output(): assert result.final_output == Foo(bar="baz") assert len(result.raw_responses) == 4, "should have four model responses" - assert len(result.to_input_list()) == 10, ( - "should have input: conversation summary, function call, function call result, message, " - "handoff, handoff output, preamble message, tool call, tool call result, final output" + assert len(result.to_input_list()) == 11, ( + "should preserve ordered history segments plus function calls, messages, handoff items, " + "and the final output without replaying the carried-forward message twice" ) - assert len(result.to_input_list(mode="normalized")) == 6, ( + assert len(result.to_input_list(mode="normalized")) == 7, ( "should have normalized replay input: conversation summary, carried-forward message, " - "preamble message, tool call, tool call result, final output" + "handoff summary, preamble message, tool call, tool call result, final output" ) assert result.last_agent == agent_1, "should have handed off to agent_1" @@ -1588,13 +1588,13 @@ async def test_streaming_events(): assert result.final_output == Foo(bar="baz") assert len(result.raw_responses) == 4, "should have four model responses" - assert len(result.to_input_list()) == 9, ( - "should have input: conversation summary, function call, function call result, message, " - "handoff, handoff output, tool call, tool call result, final output" + assert len(result.to_input_list()) == 10, ( + "should preserve ordered history segments plus function calls, messages, handoff items, " + "and the final output without replaying the carried-forward message twice" ) - assert len(result.to_input_list(mode="normalized")) == 5, ( + assert len(result.to_input_list(mode="normalized")) == 6, ( "should have normalized replay input: conversation summary, carried-forward message, " - "tool call, tool call result, final output" + "handoff summary, tool call, tool call result, final output" ) assert result.last_agent == agent_1, "should have handed off to agent_1" diff --git a/tests/test_extension_filters.py b/tests/test_extension_filters.py index a7b084e211..c86cf2c80c 100644 --- a/tests/test_extension_filters.py +++ b/tests/test_extension_filters.py @@ -333,7 +333,7 @@ def test_nest_handoff_history_wraps_transcript() -> None: nested = nest_handoff_history(data) assert isinstance(nested.input_history, tuple) - assert len(nested.input_history) == 1 + assert len(nested.input_history) == 3 summary = _as_message(nested.input_history[0]) assert summary["role"] == "assistant" summary_content = summary["content"] @@ -343,7 +343,12 @@ def test_nest_handoff_history_wraps_transcript() -> None: assert end_marker in summary_content assert "Assist reply" in summary_content assert "Hello" in summary_content + raw_message = _as_message(nested.input_history[1]) + assert "Handoff request" in str(raw_message["content"]) + handoff_summary = _as_message(nested.input_history[2]) + assert "transfer" in str(handoff_summary["content"]) assert len(nested.pre_handoff_items) == 0 + assert nested.input_items == () assert nested.new_items == data.new_items @@ -482,7 +487,7 @@ def test_nest_handoff_history_honors_custom_wrappers() -> None: try: nested = nest_handoff_history(data) assert isinstance(nested.input_history, tuple) - assert len(nested.input_history) == 1 + assert len(nested.input_history) == 2 summary = _as_message(nested.input_history[0]) summary_content = summary["content"] assert isinstance(summary_content, str) @@ -492,6 +497,7 @@ def test_nest_handoff_history_honors_custom_wrappers() -> None: ) assert lines[1].startswith("<>") assert summary_content.endswith("<>") + assert "Second reply" in str(_as_message(nested.input_history[1])["content"]) # Ensure the custom markers are parsed correctly when nesting again. second_nested = nest_handoff_history(nested) diff --git a/tests/test_handoff_history_duplication.py b/tests/test_handoff_history_duplication.py index 2a487dee38..71f265f24b 100644 --- a/tests/test_handoff_history_duplication.py +++ b/tests/test_handoff_history_duplication.py @@ -5,7 +5,9 @@ in the input sent to the next agent. """ +import dataclasses import json +from copy import deepcopy from typing import Any, cast import pytest @@ -13,11 +15,21 @@ ResponseFunctionToolCall, ResponseOutputMessage, ResponseOutputText, + ResponseToolSearchCall, + ResponseToolSearchOutputItem, ) from openai.types.responses.response_reasoning_item import ResponseReasoningItem, Summary -from agents import Agent, RunConfig, Runner, function_tool, handoff -from agents.handoffs import HandoffInputData, nest_handoff_history +from agents import Agent, RunConfig, Runner, RunState, function_tool, handoff +from agents.extensions.handoff_filters import remove_all_tools +from agents.handoffs import ( + HandoffInputData, + default_handoff_history_mapper, + nest_handoff_history, + reset_conversation_history_wrappers, + set_conversation_history_wrappers, +) +from agents.handoffs.history import _get_nested_history_owned_items from agents.items import ( HandoffCallItem, HandoffOutputItem, @@ -26,6 +38,16 @@ ToolApprovalItem, ToolCallItem, ToolCallOutputItem, + TResponseInputItem, +) +from agents.result import RunResult, RunResultStreaming +from agents.run_internal.items import ( + NestedHistoryOwnedItem, + digest_input_item, + nested_history_run_item_occurrence_key, +) +from agents.run_internal.session_persistence import ( + resolve_nested_history_owned_session_item_refs, ) from .fake_model import FakeModel @@ -92,11 +114,11 @@ def _create_handoff_output_item(agent: Agent[Any]) -> HandoffOutputItem: ) -def _create_message_item(agent: Agent) -> MessageOutputItem: +def _create_message_item(agent: Agent, *, text: str = "Hello!") -> MessageOutputItem: """Create a mock MessageOutputItem.""" raw_item = ResponseOutputMessage( id="msg_123", - content=[ResponseOutputText(text="Hello!", type="output_text", annotations=[])], + content=[ResponseOutputText(text=text, type="output_text", annotations=[])], role="assistant", status="completed", type="message", @@ -104,6 +126,19 @@ def _create_message_item(agent: Agent) -> MessageOutputItem: return MessageOutputItem(agent=agent, raw_item=raw_item, type="message_output_item") +def _input_item_text(item: TResponseInputItem) -> str: + content = item.get("content") + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + return "".join( + part.get("text", "") + for part in content + if isinstance(part, dict) and isinstance(part.get("text"), str) + ) + + def _create_reasoning_item(agent: Agent) -> ReasoningItem: """Create a mock ReasoningItem.""" raw_item = ResponseReasoningItem( @@ -228,8 +263,7 @@ def test_new_items_handoff_output_is_filtered_for_input(self): def test_message_items_are_preserved_in_new_items(self): """Verify MessageOutputItem in new_items is preserved. - Message items have a 'role' and should NOT be filtered from input_items. - Note: pre_handoff_items are converted to summary text regardless of type. + Message items have a role and should be forwarded losslessly in input_history. """ agent = _create_mock_agent() @@ -241,12 +275,117 @@ def test_message_items_are_preserved_in_new_items(self): nested = nest_handoff_history(handoff_data) - # Message items should be preserved in new_items assert len(nested.new_items) == 1, "MessageOutputItem should be preserved in new_items" - # And in input_items (since it has a role) - assert nested.input_items is not None - assert len(nested.input_items) == 1, "MessageOutputItem should be preserved in input_items" - assert isinstance(nested.input_items[0], MessageOutputItem) + assert nested.input_items == () + assert len(nested.input_history) == 2 + raw_message = cast(dict[str, Any], nested.input_history[1]) + assert raw_message["role"] == "assistant" + assert "Hello!" in str(raw_message["content"]) + + def test_forwarded_items_are_excluded_from_summary_until_the_next_handoff(self): + """A raw continuation item should have exactly one history owner at a time.""" + agent = _create_mock_agent() + message_item = _create_message_item(agent) + handoff_data = HandoffInputData( + input_history=({"role": "user", "content": "Hello"},), + pre_handoff_items=(), + new_items=(message_item,), + ) + + first_nested = nest_handoff_history(handoff_data) + + first_summary = str(cast(dict[str, Any], first_nested.input_history[0])["content"]) + assert "Hello!" not in first_summary + assert first_nested.input_items == () + assert len(first_nested.input_history) == 2 + forwarded_message = cast(dict[str, Any], first_nested.input_history[1]) + assert forwarded_message["role"] == "assistant" + assert "_agents_nested_history_token" not in forwarded_message + + second_nested = nest_handoff_history( + HandoffInputData( + input_history=first_nested.input_history, + pre_handoff_items=(), + new_items=(), + ) + ) + summary = str(cast(dict[str, Any], second_nested.input_history[0])["content"]) + assert summary.count("Hello!") == 1 + assert "_agents_nested_history_token" not in summary + assert second_nested.pre_handoff_items == () + + def test_custom_mapper_receives_clean_flattened_history(self): + """A later custom mapper must not receive SDK-only provenance metadata.""" + agent = _create_mock_agent() + first_nested = nest_handoff_history( + HandoffInputData( + input_history=({"role": "user", "content": "Hello"},), + pre_handoff_items=(), + new_items=(_create_message_item(agent),), + ) + ) + captured: list[TResponseInputItem] = [] + + def capture_mapper(transcript: list[TResponseInputItem]) -> list[TResponseInputItem]: + captured.extend(deepcopy(transcript)) + return transcript + + second_nested = nest_handoff_history( + HandoffInputData( + input_history=first_nested.input_history, + pre_handoff_items=(), + new_items=(), + ), + history_mapper=capture_mapper, + ) + + assert captured + assert all("_agents_nested_history_token" not in item for item in captured) + assert all( + "_agents_nested_history_token" not in item for item in second_nested.input_history + ) + + def test_summary_and_raw_items_preserve_chronological_order(self): + """Summary segments should not reorder interleaved lossless items.""" + agent = _create_mock_agent() + data = HandoffInputData( + input_history=({"role": "user", "content": "question"},), + pre_handoff_items=(), + new_items=( + _create_message_item(agent, text="answer"), + _create_handoff_call_item(agent), + _create_handoff_output_item(agent), + ), + ) + + nested = nest_handoff_history(data) + + assert len(nested.input_history) == 3 + prior_summary = str(cast(dict[str, Any], nested.input_history[0])["content"]) + raw_message = cast(dict[str, Any], nested.input_history[1]) + handoff_summary = str(cast(dict[str, Any], nested.input_history[2])["content"]) + assert "question" in prior_summary + assert "answer" in str(raw_message["content"]) + assert "function_call" in handoff_summary + assert "answer" not in prior_summary + assert "answer" not in handoff_summary + + def test_custom_mapper_return_value_is_exact_model_input(self): + """Nesting should not append SDK-selected items after a custom mapper.""" + agent = _create_mock_agent() + mapped_item = cast(TResponseInputItem, {"role": "user", "content": "mapped"}) + data = HandoffInputData( + input_history=({"role": "user", "content": "original"},), + pre_handoff_items=(_create_message_item(agent, text="before"),), + new_items=(_create_message_item(agent, text="after"),), + ) + + nested = nest_handoff_history(data, history_mapper=lambda _: [mapped_item]) + + assert nested.input_history == (mapped_item,) + assert nested.pre_handoff_items == () + assert nested.input_items == () + assert nested.new_items == data.new_items def test_reasoning_items_are_filtered_from_input_items(self): """Verify ReasoningItem in new_items is filtered from input_items. @@ -405,13 +544,25 @@ async def test_to_input_list_normalized_uses_filtered_continuation_after_nested_ item.get("type", "message") for item in normalized_input if isinstance(item, dict) ] - assert len(preserve_all_input) == 5 + assert len(preserve_all_input) == 6 assert "function_call" in preserve_all_types assert "function_call_output" in preserve_all_types - assert len(normalized_input) == 3 + assert sum(_input_item_text(item) == "triage summary" for item in preserve_all_input) == 1 + assert len(normalized_input) == 4 assert "function_call" not in normalized_types assert "function_call_output" not in normalized_types + replay_model = FakeModel() + replay_agent = Agent(name="replay", model=replay_model) + replay_model.add_multiple_turn_outputs([[get_text_message("replayed")]]) + replay_result = await Runner.run(replay_agent, input=preserve_all_input) + + assert replay_model.first_turn_args is not None + replay_input = replay_model.first_turn_args["input"] + assert isinstance(replay_input, list) + assert sum(_input_item_text(item) == "triage summary" for item in replay_input) == 1 + assert replay_result.final_output == "replayed" + follow_up_input = normalized_input + [{"role": "user", "content": "follow up?"}] follow_up_result = await Runner.run(delegate, input=follow_up_input) @@ -524,3 +675,1351 @@ def keep_messages_only(data: HandoffInputData) -> HandoffInputData: assert len(normalized_input) == 3 assert "function_call" not in normalized_types assert "function_call_output" not in normalized_types + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_non_nested_filtered_handoff_does_not_add_occurrence_lineage( + streamed: bool, +) -> None: + """Ordinary filtered handoffs must not mutate RunItems with nested-history metadata.""" + + def identity_filter(data: HandoffInputData) -> HandoffInputData: + return data + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=identity_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + if streamed: + streamed_result = Runner.run_streamed(first_agent, input="start") + async for _ in streamed_result.stream_events(): + pass + new_items = streamed_result.new_items + else: + result = await Runner.run(first_agent, input="start") + new_items = result.new_items + + assert all(nested_history_run_item_occurrence_key(run_item) is None for run_item in new_items) + + +@pytest.mark.asyncio +async def test_custom_filter_summary_shape_does_not_claim_equal_session_item() -> None: + """SDK-shaped custom history must not create implicit ownership by payload equality.""" + + def custom_filter(data: HandoffInputData) -> HandoffInputData: + raw_message = deepcopy(data.new_items[0].to_input_item()) + first_summary = default_handoff_history_mapper( + [cast(TResponseInputItem, {"role": "user", "content": "custom first"})] + )[0] + second_summary = default_handoff_history_mapper( + [cast(TResponseInputItem, {"role": "user", "content": "custom second"})] + )[0] + return data.clone( + input_history=(first_summary, raw_message, second_summary), + pre_handoff_items=(), + input_items=(), + ) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=custom_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "same" for item in result.to_input_list()) == 2 + + +@pytest.mark.asyncio +async def test_wrapper_reset_does_not_change_nested_history_ownership() -> None: + """Replay ownership must not depend on wrappers that are current after the run.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("handoff message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + set_conversation_history_wrappers(start="<>", end="<>") + try: + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + finally: + reset_conversation_history_wrappers() + + assert sum(_input_item_text(item) == "handoff message" for item in result.to_input_list()) == 1 + + +@pytest.mark.parametrize("chained", [False, True], ids=["direct", "chained"]) +@pytest.mark.asyncio +async def test_public_nested_history_filter_preserves_ownership(chained: bool) -> None: + """The exported nesting filter should retain provenance through built-in filter chains.""" + + def chained_filter(data: HandoffInputData) -> HandoffInputData: + return remove_all_tools(nest_handoff_history(data)) + + input_filter = chained_filter if chained else nest_handoff_history + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=input_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("handoff message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "handoff message" for item in result.to_input_list()) == 1 + + +@pytest.mark.asyncio +async def test_public_nested_history_filter_preserves_ownership_after_deepcopy() -> None: + """Copying nested input must preserve occurrence provenance for filter composition.""" + + def copied_filter(data: HandoffInputData) -> HandoffInputData: + nested = nest_handoff_history(data) + assert not isinstance(nested.input_history, str) + return nested.clone(input_history=deepcopy(nested.input_history)) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=copied_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "same" for item in result.to_input_list()) == 1 + + +@pytest.mark.asyncio +async def test_public_nested_history_filter_preserves_ownership_after_dict_rebuild() -> None: + """Rebuilding nested input mappings must preserve occurrence provenance.""" + + def rebuilt_filter(data: HandoffInputData) -> HandoffInputData: + nested = nest_handoff_history(data) + assert not isinstance(nested.input_history, str) + return nested.clone( + input_history=tuple( + cast(TResponseInputItem, dict(item)) if isinstance(item, dict) else item + for item in nested.input_history + ) + ) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=rebuilt_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "same" for item in result.to_input_list()) == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_public_nested_history_filter_rebases_owned_input_after_insertion( + streamed: bool, +) -> None: + """Inserting clean history items must not invalidate forwarded occurrence ownership.""" + + def inserting_filter(data: HandoffInputData) -> HandoffInputData: + nested = nest_handoff_history(data) + assert not isinstance(nested.input_history, str) + inserted = cast(TResponseInputItem, {"role": "user", "content": "inserted"}) + rebuilt_history = tuple( + cast(TResponseInputItem, dict(item)) if isinstance(item, dict) else item + for item in nested.input_history + ) + return nested.clone(input_history=(inserted, *rebuilt_history)) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=inserting_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + if streamed: + streamed_result = Runner.run_streamed(first_agent, input="start") + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run(first_agent, input="start") + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "same" for item in replay_input) == 1 + + +@pytest.mark.asyncio +async def test_public_nested_history_filter_data_rebuild_drops_private_ownership() -> None: + """A manually rebuilt container must not infer ownership from equal payloads.""" + + def rebuilt_filter(data: HandoffInputData) -> HandoffInputData: + nested = nest_handoff_history(data) + return HandoffInputData( + input_history=nested.input_history, + pre_handoff_items=nested.pre_handoff_items, + new_items=nested.new_items, + run_context=nested.run_context, + input_items=nested.input_items, + ) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=rebuilt_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "same" for item in result.to_input_list()) == 2 + + +@pytest.mark.asyncio +async def test_public_nested_history_filter_preserves_ownership_after_new_items_copy() -> None: + """Cloning filtered run items must preserve the private occurrence ledger.""" + + def copied_filter(data: HandoffInputData) -> HandoffInputData: + nested = nest_handoff_history(data) + return nested.clone(new_items=deepcopy(nested.new_items)) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=copied_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "same" for item in result.to_input_list()) == 1 + + +@pytest.mark.asyncio +async def test_public_nested_history_filter_does_not_own_equal_replacement() -> None: + """An equal new RunItem must remain a distinct replay occurrence.""" + + def replacement_filter(data: HandoffInputData) -> HandoffInputData: + nested = nest_handoff_history(data) + original = cast(MessageOutputItem, nested.new_items[0]) + replacement = MessageOutputItem( + agent=original.agent, + raw_item=deepcopy(original.raw_item), + ) + return nested.clone(new_items=(replacement, *nested.new_items[1:])) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=replacement_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run(first_agent, input="start") + + assert sum(_input_item_text(item) == "same" for item in result.to_input_list()) == 2 + + +def test_nested_history_private_ownership_is_not_a_dataclass_field() -> None: + """The private ownership ledger must stay out of public dataclass serialization.""" + agent = _create_mock_agent() + nested = nest_handoff_history( + HandoffInputData( + input_history="start", + pre_handoff_items=(), + new_items=(_create_message_item(agent, text="same"),), + ) + ) + + assert "_nested_history_owned_items" not in { + item_field.name for item_field in dataclasses.fields(HandoffInputData) + } + serialized = dataclasses.asdict(nested.clone(new_items=(), input_items=())) + assert "_nested_history_owned_items" not in serialized + + +def test_nested_history_input_rebase_rejects_ambiguous_equal_occurrences() -> None: + """Copied input payloads must not claim ownership when an equal occurrence is ambiguous.""" + agent = _create_mock_agent() + nested = nest_handoff_history( + HandoffInputData( + input_history="start", + pre_handoff_items=(), + new_items=(_create_message_item(agent, text="same"),), + ) + ) + assert not isinstance(nested.input_history, str) + forwarded = next( + item + for item in nested.input_history + if item.get("type") == "message" and item.get("role") == "assistant" + ) + rebuilt = nested.clone( + input_history=( + *deepcopy(nested.input_history), + deepcopy(forwarded), + ) + ) + + assert _get_nested_history_owned_items(rebuilt) == () + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_nested_history_preserves_repeated_run_item_reference_occurrences( + streamed: bool, +) -> None: + """Repeating one RunItem object must preserve both logical occurrences.""" + + def duplicate_message_filter(data: HandoffInputData) -> HandoffInputData: + message = data.new_items[0] + return nest_handoff_history(data.clone(new_items=(message, message, *data.new_items[1:]))) + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=duplicate_message_filter)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + if streamed: + streamed_result = Runner.run_streamed(first_agent, input="start") + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run(first_agent, input="start") + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "same" for item in replay_input) == 2 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_nested_history_retains_forwarded_pre_handoff_item_provenance( + streamed: bool, +) -> None: + """Lossless items from earlier turns must retain one replay occurrence.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + tool_search_call = ResponseToolSearchCall( + id="tool_search_call", + call_id="search", + arguments={"query": "profile"}, + execution="server", + status="completed", + type="tool_search_call", + ) + tool_search_output = ResponseToolSearchOutputItem( + id="tool_search_output", + call_id="search", + execution="server", + status="completed", + tools=[], + type="tool_search_output", + ) + first_model.add_multiple_turn_outputs( + [ + [tool_search_call, tool_search_output], + [get_handoff_tool_call(second_agent)], + ] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + run_result: RunResult | RunResultStreaming + + if streamed: + run_result = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in run_result.stream_events(): + pass + else: + run_result = await Runner.run(first_agent, input="start", run_config=run_config) + + replay_input = run_result.to_input_list() + replay_types = [item.get("type") for item in replay_input] + assert replay_types.count("tool_search_call") == 1 + assert replay_types.count("tool_search_output") == 1 + assert isinstance(run_result.input, list) + assert all("_agents_nested_history_token" not in item for item in run_result.input) + + owned_types = { + item_ref.run_item.type + for item_ref in run_result._nested_history_owned_session_item_refs + if item_ref.run_item is not None + } + assert {"tool_search_call_item", "tool_search_output_item"} <= owned_types + + state_json = run_result.to_state().to_json() + assert all("_agents_nested_history_token" not in item for item in state_json["original_input"]) + restored = await RunState.from_json(first_agent, state_json) + restored_owned_types = { + item_ref.run_item.type + for item_ref in restored._nested_history_owned_session_item_refs + if item_ref.run_item is not None + } + assert {"tool_search_call_item", "tool_search_output_item"} <= restored_owned_types + + +@pytest.mark.asyncio +async def test_to_input_list_during_active_stream_does_not_mutate_input() -> None: + """Inspecting an active stream must not mutate its eventual public input.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("handoff message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = Runner.run_streamed( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + input_before_inspection = deepcopy(result.input) + + assert result.to_input_list() + assert result.input == input_before_inspection + + async for _ in result.stream_events(): + pass + + assert isinstance(result.input, list) + assert all("_agents_nested_history_token" not in item for item in result.input) + assert sum(_input_item_text(item) == "handoff message" for item in result.to_input_list()) == 1 + + +@pytest.mark.asyncio +async def test_nested_history_ownership_remaps_after_new_items_insertion() -> None: + """A caller inserting a public new_items entry must not make ownership drop the new item.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("handoff message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + result.new_items.insert(0, _create_message_item(first_agent, text="inserted")) + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "inserted" for item in replay_input) == 1 + assert sum(_input_item_text(item) == "handoff message" for item in replay_input) == 1 + + +@pytest.mark.asyncio +async def test_nested_history_input_removal_does_not_claim_an_unmarked_equal_occurrence() -> None: + """An equal replacement input must not retain ownership of the removed occurrence.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + assert isinstance(result.input, list) + owned_index = result._nested_history_owned_session_item_refs[0].input_index + owned_input = result.input.pop(owned_index) + result.input.append(deepcopy(owned_input)) + + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "same" for item in replay_input) == 2 + + +@pytest.mark.asyncio +async def test_nested_history_new_item_removal_does_not_claim_an_equal_item() -> None: + """Removing the owned RunItem must not transfer ownership to an equal RunItem.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + owned_run_item = next( + item_ref.run_item + for item_ref in result._nested_history_owned_session_item_refs + if item_ref.run_item is not None + ) + equal_unowned_item = _create_message_item(first_agent, text="same") + result.new_items[:] = [equal_unowned_item] + [ + item for item in result.new_items if item is not owned_run_item + ] + + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "same" for item in replay_input) == 2 + state = result.to_state() + assert state._nested_history_owned_session_item_refs == [] + state.to_json() + + +@pytest.mark.asyncio +async def test_nested_history_ownership_survives_result_new_items_copy() -> None: + """Copying result run items must not replay a nested session occurrence twice.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + result.new_items = deepcopy(result.new_items) + + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "same" for item in replay_input) == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_nested_history_input_copy_does_not_infer_occurrence_ownership( + streamed: bool, +) -> None: + """An unmarked public-input copy must remain distinct from its session occurrence.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result: RunResult | RunResultStreaming + if streamed: + result = Runner.run_streamed( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + async for _ in result.stream_events(): + pass + else: + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + + assert isinstance(result.input, list) + result.input = deepcopy(result.input) + + replay_input = result.to_input_list() + state = result.to_state() + + assert sum(_input_item_text(item) == "same" for item in replay_input) == 2 + assert state._nested_history_owned_session_item_refs + state.to_json() + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_nested_history_input_copy_and_reorder_does_not_infer_ownership( + streamed: bool, +) -> None: + """Payload equality must not transfer ownership after a copied input reorder.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("owned once"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + + result: RunResult | RunResultStreaming + if streamed: + result = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in result.stream_events(): + pass + else: + result = await Runner.run(first_agent, input="start", run_config=run_config) + + assert isinstance(result.input, list) + result.input = list(reversed(deepcopy(result.input))) + + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "owned once" for item in replay_input) == 2 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.parametrize("mutation", ["remove", "reorder"]) +@pytest.mark.asyncio +async def test_result_input_mutation_does_not_change_state_snapshot_ownership( + streamed: bool, + mutation: str, +) -> None: + """RunState ownership must follow its saved snapshot, not the mutable result view.""" + + @function_tool(needs_approval=True) + def approval_tool() -> str: + return "approved" + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model, tools=[approval_tool]) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("owned once"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [ + [get_function_tool_call("approval_tool", "{}", call_id="approval")], + [get_text_message("done")], + ] + ) + run_config = RunConfig(nest_handoff_history=True) + + interrupted: RunResult | RunResultStreaming + if streamed: + interrupted = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in interrupted.stream_events(): + pass + else: + interrupted = await Runner.run(first_agent, input="start", run_config=run_config) + + assert len(interrupted.interruptions) == 1 + assert isinstance(interrupted.input, list) + owned_index = interrupted._nested_history_owned_session_item_refs[0].input_index + owned_item = interrupted.input[owned_index] + if mutation == "remove": + interrupted.input.pop(owned_index) + else: + target_index = 0 if owned_index != 0 else len(interrupted.input) - 1 + interrupted.input.pop(owned_index) + interrupted.input.insert(target_index, owned_item) + moved_index = next( + index for index, item in enumerate(interrupted.input) if item is owned_item + ) + assert moved_index != owned_index + + state = interrupted.to_state() + assert state._nested_history_owned_session_item_refs + state.approve(interrupted.interruptions[0]) + restored = await RunState.from_string(first_agent, state.to_string()) + + resumed: RunResult | RunResultStreaming + if streamed: + resumed = Runner.run_streamed(first_agent, restored, run_config=run_config) + async for _ in resumed.stream_events(): + pass + else: + resumed = await Runner.run(first_agent, restored, run_config=run_config) + + assert resumed.final_output == "done" + assert sum(_input_item_text(item) == "owned once" for item in resumed.to_input_list()) == 1 + + +@pytest.mark.asyncio +async def test_nested_history_ownership_revalidates_after_input_removal() -> None: + """Removing an owned input occurrence must restore its session copy during replay.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("handoff message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + assert isinstance(result.input, list) + result.input[:] = [item for item in result.input if _input_item_text(item) != "handoff message"] + + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "handoff message" for item in replay_input) == 1 + + +def test_nested_history_session_resolution_requires_exact_run_item_identity() -> None: + """Session ownership must not bind to an older equal RunItem.""" + agent = _create_mock_agent() + older = _create_message_item(agent, text="same") + owned = _create_message_item(agent, text="same") + owned_input = owned.to_input_item() + digest = digest_input_item(owned_input) + assert digest is not None + + refs = resolve_nested_history_owned_session_item_refs( + [older], + [owned_input], + [NestedHistoryOwnedItem(run_item=owned, input_index=0, digest=digest)], + ) + + assert refs == [] + + +def test_nested_history_session_resolution_keeps_surviving_owned_items() -> None: + """One missing owned RunItem must not discard another exact ownership match.""" + agent = _create_mock_agent() + kept = _create_message_item(agent, text="kept") + removed = _create_message_item(agent, text="removed") + kept_input = kept.to_input_item() + removed_input = removed.to_input_item() + kept_digest = digest_input_item(kept_input) + removed_digest = digest_input_item(removed_input) + assert kept_digest is not None + assert removed_digest is not None + + refs = resolve_nested_history_owned_session_item_refs( + [kept], + [kept_input, removed_input], + [ + NestedHistoryOwnedItem(run_item=kept, input_index=0, digest=kept_digest), + NestedHistoryOwnedItem(run_item=removed, input_index=1, digest=removed_digest), + ], + ) + + assert len(refs) == 1 + assert refs[0].run_item is kept + + +def test_nested_history_session_resolution_uses_copy_lineage_per_occurrence() -> None: + """Separately copied repeated references must resolve to separate session occurrences.""" + agent = _create_mock_agent() + repeated = _create_message_item(agent, text="same") + nested = nest_handoff_history( + HandoffInputData( + input_history="start", + pre_handoff_items=(), + new_items=(repeated, repeated), + ) + ) + assert not isinstance(nested.input_history, str) + copied_session_items = [deepcopy(repeated), deepcopy(repeated)] + + refs = resolve_nested_history_owned_session_item_refs( + copied_session_items, + nested.input_history, + _get_nested_history_owned_items(nested), + ) + + assert len(refs) == 2 + assert refs[0].run_item is copied_session_items[0] + assert refs[1].run_item is copied_session_items[1] + + +def test_nested_history_normalizes_forwarded_status_before_ownership() -> None: + """Forwarded payloads and their session digests must use canonical replay normalization.""" + agent = _create_mock_agent() + raw_message = ResponseOutputMessage.model_construct( + id="msg_none_status", + content=[ResponseOutputText(text="same", type="output_text", annotations=[])], + role="assistant", + status=None, + type="message", + ) + run_item = MessageOutputItem(agent=agent, raw_item=raw_message) + nested = nest_handoff_history( + HandoffInputData( + input_history="start", + pre_handoff_items=(), + new_items=(run_item,), + ) + ) + assert not isinstance(nested.input_history, str) + forwarded = next( + item + for item in nested.input_history + if item.get("type") == "message" and item.get("role") == "assistant" + ) + + assert "status" not in forwarded + refs = resolve_nested_history_owned_session_item_refs( + [run_item], + nested.input_history, + _get_nested_history_owned_items(nested), + ) + assert len(refs) == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_plain_handoff_preserves_prior_nested_history_ownership(streamed: bool) -> None: + """A later non-nesting handoff must not clear ownership established by an earlier handoff.""" + first_model = FakeModel() + second_model = FakeModel() + final_model = FakeModel() + final_agent = Agent(name="final", model=final_model) + second_agent = Agent( + name="second", + model=second_model, + handoffs=[handoff(final_agent, nest_handoff_history=False)], + ) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("first message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [[get_text_message("second message"), get_handoff_tool_call(final_agent)]] + ) + final_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + + if streamed: + streamed_result = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run(first_agent, input="start", run_config=run_config) + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "first message" for item in replay_input) == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_non_nested_copying_filter_preserves_prior_nested_history_ownership( + streamed: bool, +) -> None: + """A later non-nesting filter may copy history without losing prior ownership.""" + + def copy_history(data: HandoffInputData) -> HandoffInputData: + if isinstance(data.input_history, str): + return data + return data.clone(input_history=deepcopy(data.input_history)) + + first_model = FakeModel() + second_model = FakeModel() + final_model = FakeModel() + final_agent = Agent(name="final", model=final_model) + second_agent = Agent( + name="second", + model=second_model, + handoffs=[ + handoff( + final_agent, + input_filter=copy_history, + nest_handoff_history=False, + ) + ], + ) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("first message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [[get_text_message("second message"), get_handoff_tool_call(final_agent)]] + ) + final_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + + if streamed: + streamed_result = Runner.run_streamed( + first_agent, + input="start", + run_config=run_config, + ) + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run(first_agent, input="start", run_config=run_config) + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "first message" for item in replay_input) == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_copied_custom_input_items_keep_session_occurrence_for_later_nesting( + streamed: bool, +) -> None: + """Copied model items must still resolve to their exact session occurrences.""" + + def copy_model_items(data: HandoffInputData) -> HandoffInputData: + return data.clone(input_items=deepcopy(data.new_items)) + + first_model = FakeModel() + second_model = FakeModel() + final_model = FakeModel() + final_agent = Agent(name="final", model=final_model) + second_agent = Agent(name="second", model=second_model, handoffs=[final_agent]) + first_agent = Agent( + name="first", + model=first_model, + handoffs=[handoff(second_agent, input_filter=copy_model_items)], + ) + first_model.add_multiple_turn_outputs( + [[get_text_message("copied once"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_handoff_tool_call(final_agent)]]) + final_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + + if streamed: + streamed_result = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run(first_agent, input="start", run_config=run_config) + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "copied once" for item in replay_input) == 1 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_identity_filter_preserves_prior_nested_history_ownership(streamed: bool) -> None: + """A later no-op filter must retain ownership represented in the filtered input.""" + + def identity_filter(data: HandoffInputData) -> HandoffInputData: + return data + + first_model = FakeModel() + second_model = FakeModel() + final_model = FakeModel() + final_agent = Agent(name="final", model=final_model) + second_agent = Agent( + name="second", + model=second_model, + handoffs=[handoff(final_agent, input_filter=identity_filter)], + ) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + first_model.add_multiple_turn_outputs( + [[get_text_message("first message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [[get_text_message("second message"), get_handoff_tool_call(final_agent)]] + ) + final_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + + if streamed: + streamed_result = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run(first_agent, input="start", run_config=run_config) + replay_input = result.to_input_list() + + assert sum(_input_item_text(item) == "first message" for item in replay_input) == 1 + + +@pytest.mark.parametrize("streamed", [False, True]) +@pytest.mark.asyncio +async def test_nested_handoff_history_preserves_identical_messages_across_turns( + streamed: bool, +) -> None: + """Raw-tail ownership should preserve equal messages from distinct logical turns.""" + + @function_tool + def continue_work() -> str: + return "continue" + + first_model = FakeModel() + second_model = FakeModel() + final_model = FakeModel() + final_agent = Agent(name="final", model=final_model) + second_agent = Agent( + name="second", + model=second_model, + tools=[continue_work], + handoffs=[final_agent], + ) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + + first_model.add_multiple_turn_outputs( + [[get_text_message("same"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [ + [get_text_message("same"), get_function_tool_call("continue_work", "{}")], + [get_text_message("same"), get_handoff_tool_call(final_agent)], + ] + ) + final_model.add_multiple_turn_outputs([[get_text_message("same")]]) + + if streamed: + streamed_result = Runner.run_streamed( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + async for _ in streamed_result.stream_events(): + pass + replay_input = streamed_result.to_input_list() + else: + result = await Runner.run( + first_agent, + input="start", + run_config=RunConfig(nest_handoff_history=True), + ) + replay_input = result.to_input_list() + + final_input = final_model.last_turn_args["input"] + summary = str(cast(dict[str, Any], final_input[0])["content"]) + assert summary.count("same") == 2 + assert sum(_input_item_text(item) == "same" for item in replay_input) == 4 + + +@pytest.mark.asyncio +async def test_explicit_default_handoff_history_mapper_is_honored() -> None: + """An explicitly configured mapper should own the exact model input.""" + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + + first_model.add_multiple_turn_outputs( + [[get_text_message("handoff message"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + await Runner.run( + first_agent, + input="start", + run_config=RunConfig( + nest_handoff_history=True, + handoff_history_mapper=default_handoff_history_mapper, + ), + ) + + assert second_model.first_turn_args is not None + second_input = second_model.first_turn_args["input"] + assert isinstance(second_input, list) + assert len(second_input) == 1 + summary = str(cast(dict[str, Any], second_input[0])["content"]) + assert "handoff message" in summary + assert "transfer_to_second" in summary + + +@pytest.mark.asyncio +async def test_nested_handoff_history_partition_survives_interruption_resume() -> None: + """The summary/raw partition should survive RunState serialization.""" + + @function_tool(needs_approval=True) + def approval_tool() -> str: + return "approved" + + first_model = FakeModel() + second_model = FakeModel() + final_model = FakeModel() + final_agent = Agent(name="final", model=final_model) + second_agent = Agent( + name="second", + model=second_model, + tools=[approval_tool], + handoffs=[final_agent], + ) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + run_config = RunConfig(nest_handoff_history=True) + + first_model.add_multiple_turn_outputs( + [[get_text_message("once"), get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [ + [get_function_tool_call("approval_tool", "{}", call_id="approval")], + [get_text_message("once"), get_handoff_tool_call(final_agent)], + ] + ) + final_model.add_multiple_turn_outputs([[get_text_message("done")]]) + + interrupted = await Runner.run(first_agent, input="start", run_config=run_config) + assert len(interrupted.interruptions) == 1 + assert sum(_input_item_text(item) == "once" for item in interrupted.to_input_list()) == 1 + state = interrupted.to_state() + assert state._nested_history_owned_session_item_refs + serialized_refs = state.to_json()["nested_history_owned_session_item_refs"] + assert all( + isinstance(item_ref, dict) + and isinstance(item_ref.get("digest"), str) + and len(item_ref["digest"]) == 64 + for item_ref in serialized_refs + ) + state.approve(interrupted.interruptions[0]) + serialized_state = state.to_string() + restored = await RunState.from_string(first_agent, serialized_state) + assert restored._nested_history_owned_session_item_refs == ( + state._nested_history_owned_session_item_refs + ) + + resumed = await Runner.run(first_agent, restored, run_config=run_config) + + assert resumed.final_output == "done" + final_input = final_model.last_turn_args["input"] + summary = str(cast(dict[str, Any], final_input[0])["content"]) + assert summary.count("once") == 1 + assert sum(_input_item_text(item) == "once" for item in resumed.to_input_list()) == 2 + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_nested_history_resume_to_final_preserves_status_less_ownership( + streamed: bool, +) -> None: + """Resume without another handoff must retain one status-less nested occurrence.""" + + @function_tool(needs_approval=True) + def approval_tool() -> str: + return "approved" + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model, tools=[approval_tool]) + first_agent = Agent(name="first", model=first_model, handoffs=[second_agent]) + run_config = RunConfig(nest_handoff_history=True) + status_less_message = ResponseOutputMessage.model_construct( + id="msg_none_status", + content=[ResponseOutputText(text="once", type="output_text", annotations=[])], + role="assistant", + status=None, + type="message", + ) + first_model.add_multiple_turn_outputs( + [[status_less_message, get_handoff_tool_call(second_agent)]] + ) + second_model.add_multiple_turn_outputs( + [ + [get_function_tool_call("approval_tool", "{}", call_id="approval")], + [get_text_message("done")], + ] + ) + + if streamed: + interrupted_stream = Runner.run_streamed( + first_agent, + input="start", + run_config=run_config, + ) + async for _ in interrupted_stream.stream_events(): + pass + interruptions = interrupted_stream.interruptions + state = interrupted_stream.to_state() + else: + interrupted_result = await Runner.run( + first_agent, + input="start", + run_config=run_config, + ) + interruptions = interrupted_result.interruptions + state = interrupted_result.to_state() + + assert len(interruptions) == 1 + state.approve(interruptions[0]) + restored = await RunState.from_string(first_agent, state.to_string()) + + if streamed: + resumed_stream = Runner.run_streamed(first_agent, restored, run_config=run_config) + async for _ in resumed_stream.stream_events(): + pass + final_output = resumed_stream.final_output + replay_input = resumed_stream.to_input_list() + else: + resumed_result = await Runner.run(first_agent, restored, run_config=run_config) + final_output = resumed_result.final_output + replay_input = resumed_result.to_input_list() + + assert final_output == "done" + assert sum(_input_item_text(item) == "once" for item in replay_input) == 1 + assert all("_agents_nested_history_token" not in item for item in replay_input) + assert second_model.last_turn_args is not None + assert all( + "_agents_nested_history_token" not in item for item in second_model.last_turn_args["input"] + ) + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non_streamed", "streamed"]) +@pytest.mark.parametrize( + "legacy_snapshot", + [False, True], + ids=["current_schema", "schema_1_12"], +) +@pytest.mark.asyncio +async def test_first_nested_handoff_after_restore_uses_explicit_occurrence_lineage( + streamed: bool, + legacy_snapshot: bool, +) -> None: + """Current snapshots restore lineage while legacy snapshots remain conservative.""" + + @function_tool(needs_approval=True) + def approval_tool() -> str: + return "approved" + + first_model = FakeModel() + second_model = FakeModel() + second_agent = Agent(name="second", model=second_model) + first_agent = Agent( + name="first", + model=first_model, + tools=[approval_tool], + handoffs=[second_agent], + ) + tool_search_call = ResponseToolSearchCall( + id="tool_search_call", + call_id="search", + arguments={"query": "profile"}, + execution="server", + status="completed", + type="tool_search_call", + ) + tool_search_output = ResponseToolSearchOutputItem( + id="tool_search_output", + call_id="search", + execution="server", + status="completed", + tools=[], + type="tool_search_output", + ) + first_model.add_multiple_turn_outputs( + [ + [ + tool_search_call, + tool_search_output, + get_function_tool_call("approval_tool", "{}", call_id="approval"), + ], + [get_handoff_tool_call(second_agent)], + ] + ) + second_model.add_multiple_turn_outputs([[get_text_message("done")]]) + run_config = RunConfig(nest_handoff_history=True) + interrupted: RunResult | RunResultStreaming + + if streamed: + interrupted = Runner.run_streamed(first_agent, input="start", run_config=run_config) + async for _ in interrupted.stream_events(): + pass + else: + interrupted = await Runner.run(first_agent, input="start", run_config=run_config) + + assert len(interrupted.interruptions) == 1 + state = interrupted.to_state() + state.approve(interrupted.interruptions[0]) + state._session_items.insert(0, _create_message_item(first_agent, text="session only")) + state_json = state.to_json() + if legacy_snapshot: + state_json["$schemaVersion"] = "1.12" + state_json.pop("nested_history_owned_session_item_refs") + state_json.pop("generated_session_item_indexes") + restored = await RunState.from_json(first_agent, state_json) + resumed: RunResult | RunResultStreaming + + if streamed: + resumed = Runner.run_streamed(first_agent, restored, run_config=run_config) + async for _ in resumed.stream_events(): + pass + else: + resumed = await Runner.run(first_agent, restored, run_config=run_config) + + replay_types = [item.get("type") for item in resumed.to_input_list()] + expected_occurrences = 2 if legacy_snapshot else 1 + assert replay_types.count("tool_search_call") == expected_occurrences + assert replay_types.count("tool_search_output") == expected_occurrences diff --git a/tests/test_run_internal_items.py b/tests/test_run_internal_items.py index c7997930d1..721787b011 100644 --- a/tests/test_run_internal_items.py +++ b/tests/test_run_internal_items.py @@ -1,5 +1,6 @@ from __future__ import annotations +import dataclasses from typing import Any, cast import pytest @@ -364,6 +365,122 @@ def _raise_json_error(*_args: Any, **_kwargs: Any) -> str: assert run_items.fingerprint_input_item({"type": "message", "role": "user"}) is None +def test_digest_input_item_is_fixed_size_and_content_opaque() -> None: + item = cast( + TResponseInputItem, + {"type": "message", "role": "assistant", "content": "sensitive-history-content"}, + ) + + digest = run_items.digest_input_item(item) + + assert digest is not None + assert len(digest) == 64 + assert "sensitive-history-content" not in digest + + +def test_nested_history_digest_treats_default_assistant_status_as_equivalent() -> None: + without_status = cast( + TResponseInputItem, + {"type": "message", "role": "assistant", "content": "same"}, + ) + with_completed_status = cast( + TResponseInputItem, + { + "type": "message", + "role": "assistant", + "content": "same", + "status": "completed", + }, + ) + + assert run_items.digest_input_item(without_status) == run_items.digest_input_item( + with_completed_status + ) + + +def test_resumed_input_normalizes_clean_nested_history_items() -> None: + item = cast(TResponseInputItem, {"role": "assistant", "content": "same"}) + + resumed = run_items.normalize_resumed_input([item]) + + assert isinstance(resumed, list) + assert resumed == [item] + + +def test_filter_nested_history_owned_refs_requires_the_exact_input_occurrence() -> None: + item = cast(TResponseInputItem, {"role": "assistant", "content": "same"}) + equal_item = cast(TResponseInputItem, dict(item)) + digest = run_items.digest_input_item(item) + assert digest is not None + item_ref = run_items.NestedHistoryOwnedItemRef( + session_index=1, + digest=digest, + input_index=0, + input_item=item, + ) + + assert run_items.filter_nested_history_owned_item_refs_for_input([equal_item], [item_ref]) == [] + assert run_items.filter_nested_history_owned_item_refs_for_input( + [equal_item, item], + [item_ref], + ) == [dataclasses.replace(item_ref, input_index=1)] + + +def test_reconcile_nested_history_rewrite_rejects_ambiguous_equal_occurrences() -> None: + item = cast(TResponseInputItem, {"role": "assistant", "content": "same"}) + equal_item = cast(TResponseInputItem, dict(item)) + digest = run_items.digest_input_item(item) + assert digest is not None + item_ref = run_items.NestedHistoryOwnedItemRef( + session_index=0, + digest=digest, + input_index=0, + input_item=item, + ) + + rewritten, retained = run_items.reconcile_nested_history_owned_input_after_rewrite( + [item, equal_item], + [ + cast(TResponseInputItem, dict(item)), + cast(TResponseInputItem, dict(equal_item)), + ], + [item_ref], + ) + + assert retained == [] + assert rewritten == [item, equal_item] + + +def test_reconcile_nested_history_rewrite_keeps_fully_owned_equal_occurrences() -> None: + items = [ + cast(TResponseInputItem, {"role": "assistant", "content": "same"}), + cast(TResponseInputItem, {"role": "assistant", "content": "same"}), + ] + digest = run_items.digest_input_item(items[0]) + assert digest is not None + item_refs = [ + run_items.NestedHistoryOwnedItemRef( + session_index=index, + digest=digest, + input_index=index, + input_item=items[index], + ) + for index in range(2) + ] + + rewritten, retained = run_items.reconcile_nested_history_owned_input_after_rewrite( + items, + [ + cast(TResponseInputItem, dict(items[0])), + cast(TResponseInputItem, dict(items[1])), + ], + item_refs, + ) + + assert retained == item_refs + assert [item_ref.input_item for item_ref in retained] == rewritten + + def test_strip_metadata_and_reasoning_id_helpers_keep_non_matching_items() -> None: raw = cast(TResponseInputItem, "raw-item") non_reasoning = cast(TResponseInputItem, {"type": "message", "id": "msg_1"}) diff --git a/tests/test_run_state.py b/tests/test_run_state.py index a6955a777b..12ac9ca1d2 100644 --- a/tests/test_run_state.py +++ b/tests/test_run_state.py @@ -7,6 +7,7 @@ import json import logging from collections.abc import AsyncIterator, Callable, Mapping +from copy import deepcopy from dataclasses import dataclass from datetime import datetime from pathlib import Path @@ -60,7 +61,13 @@ ) from agents.run_context import RunContextWrapper from agents.run_internal.agent_runner_helpers import resolve_trace_settings -from agents.run_internal.items import run_items_to_input_items +from agents.run_internal.items import ( + NestedHistoryOwnedItemRef, + digest_input_item, + ensure_nested_history_run_item_occurrence_key, + run_item_to_input_item, + run_items_to_input_items, +) from agents.run_internal.run_loop import ( NextStepInterruption, ProcessedResponse, @@ -1810,7 +1817,7 @@ async def test_preserves_usage_data(self): serialized = json.loads(str_data) new_state = await RunState.from_string(agent, str_data) - assert serialized["$schemaVersion"] == "1.12" + assert serialized["$schemaVersion"] == "1.13" assert serialized["context"]["usage"]["input_tokens_details"] == [ {"cached_tokens": 3, "cache_write_tokens": 7} ] @@ -4989,6 +4996,7 @@ def test_supported_schema_versions_match_released_boundary(self): "1.9", "1.10", "1.11", + "1.12", CURRENT_SCHEMA_VERSION, } ) @@ -4999,6 +5007,178 @@ def test_supported_schema_versions_have_non_empty_summaries(self): assert CURRENT_SCHEMA_VERSION in SCHEMA_VERSION_SUMMARIES assert all(summary.strip() for summary in SCHEMA_VERSION_SUMMARIES.values()) + @pytest.mark.asyncio + async def test_nested_history_ownership_round_trips_and_defaults_for_schema_1_12(self): + """New snapshots persist ownership while released 1.12 snapshots default safely.""" + agent = Agent(name="TestAgent") + message_item = MessageOutputItem(agent=agent, raw_item=make_message_output(text="owned")) + input_item = run_item_to_input_item(message_item) + assert input_item is not None + digest = digest_input_item(input_item) + assert digest is not None + state: RunState[Any] = make_state( + agent, + context=RunContextWrapper(context={}), + original_input=[input_item], + ) + state._session_items = [message_item] + state._generated_items = [message_item] + item_ref = NestedHistoryOwnedItemRef( + session_index=0, + digest=digest, + input_index=0, + run_item=message_item, + input_item=input_item, + ) + state._nested_history_owned_session_item_refs = [item_ref] + + serialized = state.to_json() + restored = await RunState.from_json(agent, serialized) + + assert serialized["nested_history_owned_session_item_refs"] == [ + { + "index": 0, + "digest": digest, + "input_index": 0, + } + ] + assert serialized["generated_session_item_indexes"] == [0] + assert restored._nested_history_owned_session_item_refs == [item_ref] + assert restored._generated_items[0] is restored._session_items[0] + assert isinstance(restored._original_input, list) + assert ( + restored._nested_history_owned_session_item_refs[0].input_item + is (restored._original_input[0]) + ) + + serialized["$schemaVersion"] = "1.12" + serialized.pop("nested_history_owned_session_item_refs") + serialized.pop("generated_session_item_indexes") + restored_1_12 = await RunState.from_json(agent, serialized) + + assert restored_1_12._nested_history_owned_session_item_refs == [] + assert restored_1_12._generated_items[0] is not restored_1_12._session_items[0] + + @pytest.mark.asyncio + async def test_nested_history_ownership_remaps_after_skipped_session_item(self): + """A skipped unrelated item must not shift a surviving ownership reference.""" + agent = Agent(name="TestAgent") + skipped_item = MessageOutputItem( + agent=agent, + raw_item=make_message_output(text="skip me"), + ) + owned_item = MessageOutputItem( + agent=agent, + raw_item=make_message_output(text="owned"), + ) + owned_input = run_item_to_input_item(owned_item) + assert owned_input is not None + digest = digest_input_item(owned_input) + assert digest is not None + state: RunState[Any] = make_state( + agent, + context=RunContextWrapper(context={}), + original_input=[owned_input], + ) + state._generated_items = [owned_item] + state._session_items = [skipped_item, owned_item] + state._nested_history_owned_session_item_refs = [ + NestedHistoryOwnedItemRef( + session_index=1, + digest=digest, + input_index=0, + run_item=owned_item, + input_item=owned_input, + ) + ] + serialized = state.to_json() + serialized["session_items"][0]["agent"]["name"] = "UnknownAgent" + + restored = await RunState.from_json(agent, serialized) + + assert len(restored._session_items) == 1 + assert restored._generated_items[0] is restored._session_items[0] + assert restored._nested_history_owned_session_item_refs[0].session_index == 0 + assert ( + restored._nested_history_owned_session_item_refs[0].run_item + is (restored._session_items[0]) + ) + + @pytest.mark.asyncio + async def test_copied_generated_item_round_trips_to_its_session_occurrence(self): + """The generated/session sidecar must recognize an explicitly copied occurrence.""" + agent = Agent(name="TestAgent") + session_item = MessageOutputItem( + agent=agent, + raw_item=make_message_output(text="copied"), + ) + ensure_nested_history_run_item_occurrence_key(session_item) + generated_copy = deepcopy(session_item) + state: RunState[Any] = make_state( + agent, + context=RunContextWrapper(context={}), + ) + state._generated_items = [generated_copy] + state._session_items = [session_item] + + serialized = state.to_json() + restored = await RunState.from_json(agent, serialized) + + assert serialized["generated_session_item_indexes"] == [0] + assert "_agents_nested_history_occurrence_key" not in json.dumps(serialized) + assert restored._generated_items[0] is restored._session_items[0] + + @pytest.mark.asyncio + async def test_equal_generated_replacement_does_not_claim_session_occurrence(self): + """Equal payloads without explicit lineage must serialize as distinct occurrences.""" + agent = Agent(name="TestAgent") + session_item = MessageOutputItem( + agent=agent, + raw_item=make_message_output(text="same"), + ) + generated_replacement = MessageOutputItem( + agent=agent, + raw_item=deepcopy(session_item.raw_item), + ) + state: RunState[Any] = make_state( + agent, + context=RunContextWrapper(context={}), + ) + state._generated_items = [generated_replacement] + state._session_items = [session_item] + + serialized = state.to_json() + restored = await RunState.from_json(agent, serialized) + + assert serialized["generated_session_item_indexes"] == [None] + assert restored._generated_items[0] is not restored._session_items[0] + + serialized["$schemaVersion"] = "1.12" + serialized.pop("nested_history_owned_session_item_refs") + serialized.pop("generated_session_item_indexes") + restored_1_12 = await RunState.from_json(agent, serialized) + + assert restored_1_12._generated_items[0] is not restored_1_12._session_items[0] + + @pytest.mark.asyncio + async def test_ambiguous_copied_generated_item_does_not_claim_equal_session_occurrence(self): + """An equal partial copy must remain separate when its session occurrence is ambiguous.""" + agent = Agent(name="TestAgent") + first = MessageOutputItem(agent=agent, raw_item=make_message_output(text="same")) + second = MessageOutputItem(agent=agent, raw_item=make_message_output(text="same")) + state: RunState[Any] = make_state( + agent, + context=RunContextWrapper(context={}), + ) + state._generated_items = [deepcopy(second)] + state._session_items = [first, second] + + serialized = state.to_json() + restored = await RunState.from_json(agent, serialized) + + assert serialized["generated_session_item_indexes"] == [None] + assert all(restored._generated_items[0] is not item for item in restored._session_items) + @pytest.mark.asyncio async def test_from_json_accepts_schema_version_1_5_without_sandbox_payload(self): """RunState snapshots written before sandbox resume support should still restore."""