From 0a12daa5de5d7363164c688a1ab4050f5498158d Mon Sep 17 00:00:00 2001 From: Kazuhiro Sera Date: Fri, 14 Aug 2026 10:40:34 +0900 Subject: [PATCH] fix: detach RunState interruption snapshots --- src/agents/items.py | 10 +- src/agents/run_state.py | 674 +++++++- tests/test_run_state.py | 1788 +++++++++++++++++++++- tests/test_tool_name_collision_policy.py | 33 +- 4 files changed, 2437 insertions(+), 68 deletions(-) diff --git a/src/agents/items.py b/src/agents/items.py index f3d2d1a464..a4da32fe97 100644 --- a/src/agents/items.py +++ b/src/agents/items.py @@ -12,7 +12,9 @@ from openai.types.responses import ( Response, ResponseComputerToolCall, + ResponseCustomToolCall, ResponseFileSearchToolCall, + ResponseFunctionShellToolCall, ResponseFunctionShellToolCallOutput, ResponseFunctionToolCall, ResponseFunctionWebSearch, @@ -547,7 +549,13 @@ def to_input_item(self) -> TResponseInputItem: # Union type for tool approval raw items - supports function tools, hosted tools, shell tools, etc. ToolApprovalRawItem: TypeAlias = ( - ResponseFunctionToolCall | McpCall | McpApprovalRequest | LocalShellCall | dict[str, Any] + ResponseFunctionToolCall + | ResponseCustomToolCall + | ResponseFunctionShellToolCall + | McpCall + | McpApprovalRequest + | LocalShellCall + | dict[str, Any] ) diff --git a/src/agents/run_state.py b/src/agents/run_state.py index 01b1cfb5b5..c8881b3423 100644 --- a/src/agents/run_state.py +++ b/src/agents/run_state.py @@ -6,12 +6,13 @@ import copy import dataclasses import json +import math import threading from collections import deque from collections.abc import Callable, Collection, Iterator, Mapping, Sequence from dataclasses import dataclass, field from pathlib import Path -from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal, cast +from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal, cast, get_args from uuid import uuid4 from openai.types.responses import ( @@ -36,7 +37,7 @@ Program, ProgramOutput, ) -from pydantic import StringConstraints, TypeAdapter, ValidationError +from pydantic import BaseModel, StringConstraints, TypeAdapter, ValidationError from typing_extensions import TypedDict, TypeVar from ._tool_identity import ( @@ -86,6 +87,7 @@ ReasoningItem, RunItem, ToolApprovalItem, + ToolApprovalRawItem, ToolCallItem, ToolCallOutputItem, ToolSearchCallItem, @@ -244,6 +246,18 @@ class _LocalShellCallOutputPayload(TypedDict): _MCP_APPROVAL_RESPONSE_ADAPTER: TypeAdapter[McpApprovalResponse] = TypeAdapter(McpApprovalResponse) _HANDOFF_OUTPUT_ADAPTER: TypeAdapter[TResponseInputItem] = TypeAdapter(TResponseInputItem) _LOCAL_SHELL_CALL_ADAPTER: TypeAdapter[LocalShellCall] = TypeAdapter(LocalShellCall) +_TOOL_APPROVAL_MODEL_TYPES: tuple[type[BaseModel], ...] = tuple( + raw_item_type + for raw_item_type in get_args(ToolApprovalRawItem) + if isinstance(raw_item_type, type) and issubclass(raw_item_type, BaseModel) +) +_TOOL_APPROVAL_MODEL_ADAPTERS: tuple[tuple[type[BaseModel], TypeAdapter[Any]], ...] = tuple( + (model_type, TypeAdapter(model_type)) for model_type in _TOOL_APPROVAL_MODEL_TYPES +) +_UNSAFE_PYDANTIC_SUBTYPE_HOOKS = frozenset({"__getattr__", "__getattribute__"}) +_PYDANTIC_PUBLIC_COPY_INSTANCE_ATTRIBUTES = frozenset( + {"__dict__", "__pydantic_extra__", "__pydantic_fields_set__"} +) _MISSING_CONTEXT_SENTINEL = object() _ALLOWED_MISSING_MESSAGE_FIELDS = frozenset({"status"}) @@ -253,6 +267,479 @@ def _deserialize_tool_origin(data: Any) -> ToolOrigin | None: return ToolOrigin.from_json_dict(data) +def _static_type_mro(value: Any) -> tuple[type[Any], ...]: + """Return an instance's real MRO without consulting instance attributes.""" + return cast(tuple[type[Any], ...], type.__getattribute__(type(value), "__mro__")) + + +def _declared_model_type_from_annotation( + annotation: Any, + value_mro: tuple[type[Any], ...], +) -> type[BaseModel] | None: + """Resolve a nested model from trusted Pydantic field annotation identities.""" + pending = [annotation] + visited: set[int] = set() + while pending: + candidate = pending.pop() + candidate_id = id(candidate) + if candidate_id in visited: + continue + visited.add(candidate_id) + if isinstance(candidate, type): + candidate_mro = type.__getattribute__(candidate, "__mro__") + if BaseModel in candidate_mro and candidate in value_mro: + return cast(type[BaseModel], candidate) + pending.extend(get_args(candidate)) + return None + + +def _copy_json_compatible_value(value: Any, active_container_ids: set[int]) -> Any: + """Copy bounded JSON-shaped data without invoking container or model hooks.""" + if value is None or type(value) is bool: + return value + value_mro = _static_type_mro(value) + if str in value_mro: + return str.__str__(value) + if int in value_mro: + return int.__int__(value) + if float in value_mro: + copied_float = float.__float__(value) + if not math.isfinite(copied_float): + raise TypeError("Non-finite number in tool approval payload") + return copied_float + value_id = id(value) + if value_id in active_container_ids: + raise TypeError("Cyclic tool approval payload") + if dict in value_mro: + active_container_ids.add(value_id) + try: + copied_dict: dict[str, Any] = {} + for key, item in dict.items(value): + if str not in _static_type_mro(key): + raise TypeError("Non-string key in tool approval payload") + normalized_key = str.__str__(key) + if normalized_key in copied_dict: + raise TypeError("Colliding key in tool approval payload") + copied_dict[normalized_key] = _copy_json_compatible_value( + item, + active_container_ids, + ) + return copied_dict + finally: + active_container_ids.remove(value_id) + if list in value_mro: + active_container_ids.add(value_id) + try: + return [ + _copy_json_compatible_value(item, active_container_ids) + for item in list.__iter__(value) + ] + finally: + active_container_ids.remove(value_id) + if tuple in value_mro: + active_container_ids.add(value_id) + try: + return [ + _copy_json_compatible_value(item, active_container_ids) + for item in tuple.__iter__(value) + ] + finally: + active_container_ids.remove(value_id) + raise TypeError("Unsupported value in tool approval payload") + + +def _copy_pydantic_value( + value: Any, + active_container_ids: set[int], + *, + allow_models: bool, + declared_model_type: type[BaseModel] | None = None, + declared_annotation: Any = None, +) -> Any: + """Copy a Pydantic value before public serialization can traverse untrusted data.""" + if value is None or type(value) is bool: + return value + value_mro = _static_type_mro(value) + if str in value_mro: + return str.__str__(value) + if int in value_mro: + return int.__int__(value) + if float in value_mro: + copied_float = float.__float__(value) + if not math.isfinite(copied_float): + raise TypeError("Non-finite number in tool approval payload") + return copied_float + + value_id = id(value) + if value_id in active_container_ids: + raise TypeError("Cyclic tool approval payload") + + if BaseModel in value_mro: + if not allow_models: + raise TypeError("Unsupported model in tool approval metadata") + if declared_model_type is None: + declared_model_type = _declared_model_type_from_annotation( + declared_annotation, + value_mro, + ) + if declared_model_type is None or declared_model_type not in value_mro: + raise TypeError("Unsupported model in tool approval payload") + trusted_model_mro = frozenset(type.__getattribute__(declared_model_type, "__mro__")) + for subtype in value_mro: + if subtype in trusted_model_mro: + continue + subtype_namespace = type.__getattribute__(subtype, "__dict__") + if _UNSAFE_PYDANTIC_SUBTYPE_HOOKS & subtype_namespace.keys(): + raise TypeError("Unsupported model hooks in tool approval payload") + for attribute_name in _PYDANTIC_PUBLIC_COPY_INSTANCE_ATTRIBUTES: + for attribute_owner in value_mro: + owner_namespace = type.__getattribute__(attribute_owner, "__dict__") + if attribute_name not in owner_namespace: + continue + if attribute_owner not in trusted_model_mro: + raise TypeError("Unsupported model storage hooks in tool approval payload") + break + active_container_ids.add(value_id) + try: + declared_fields = declared_model_type.model_fields + model_storage = object.__getattribute__(value, "__dict__") + if type(model_storage) is not dict: + raise TypeError("Unsupported tool approval model storage") + model_extra = BaseModel.model_extra.__get__(value, BaseModel) + copied_extra: dict[str, Any] = {} + if model_extra is not None: + if type(model_extra) is not dict: + raise TypeError("Unsupported tool approval model extras") + seen_extra_names = set(declared_fields) + for extra_name, extra_value in dict.items(model_extra): + if str not in _static_type_mro(extra_name): + raise TypeError("Non-string key in tool approval model extras") + normalized_name = str.__str__(extra_name) + if normalized_name in seen_extra_names: + raise TypeError("Colliding key in tool approval model extras") + seen_extra_names.add(normalized_name) + copied_extra[normalized_name] = _copy_pydantic_value( + extra_value, + active_container_ids, + allow_models=False, + ) + + copied_fields: dict[str, Any] = {} + for field_name, field_value in dict.items(model_storage): + if str not in _static_type_mro(field_name): + raise TypeError("Non-string field name in tool approval payload") + normalized_field_name = str.__str__(field_name) + if normalized_field_name not in declared_fields: + continue + if normalized_field_name in copied_fields: + raise TypeError("Colliding field name in tool approval payload") + copied_fields[normalized_field_name] = _copy_pydantic_value( + field_value, + active_container_ids, + allow_models=True, + declared_annotation=declared_fields[normalized_field_name].annotation, + ) + + source_fields_set = BaseModel.model_fields_set.__get__(value, BaseModel) + if type(source_fields_set) is not set: + raise TypeError("Unsupported tool approval model fields set") + copied_fields_set: set[str] = set() + allowed_fields_set = set(declared_fields) | set(copied_extra) + for field_name in set.__iter__(source_fields_set): + if str not in _static_type_mro(field_name): + raise TypeError("Non-string field name in tool approval fields set") + normalized_field_name = str.__str__(field_name) + if normalized_field_name in copied_fields_set: + raise TypeError("Colliding field name in tool approval fields set") + if normalized_field_name in allowed_fields_set: + copied_fields_set.add(normalized_field_name) + + copied_model = declared_model_type.model_construct( + _fields_set=set(copied_fields_set), + **copied_fields, + **copied_extra, + ) + constructed_fields_set = BaseModel.model_fields_set.__get__( + copied_model, + BaseModel, + ) + set.clear(constructed_fields_set) + set.update(constructed_fields_set, copied_fields_set) + return copied_model + finally: + active_container_ids.remove(value_id) + + if dict in value_mro: + active_container_ids.add(value_id) + try: + copied_dict: dict[str, Any] = {} + seen_names: set[str] = set() + for key, item in dict.items(value): + if str not in _static_type_mro(key): + raise TypeError("Non-string key in tool approval payload") + normalized_key = str.__str__(key) + if normalized_key in seen_names: + raise TypeError("Colliding key in tool approval payload") + seen_names.add(normalized_key) + copied_dict[normalized_key] = _copy_pydantic_value( + item, + active_container_ids, + allow_models=allow_models, + declared_annotation=declared_annotation, + ) + return copied_dict + finally: + active_container_ids.remove(value_id) + + if list in value_mro: + active_container_ids.add(value_id) + try: + return [ + _copy_pydantic_value( + item, + active_container_ids, + allow_models=allow_models, + declared_annotation=declared_annotation, + ) + for item in list.__iter__(value) + ] + finally: + active_container_ids.remove(value_id) + + if tuple in value_mro: + active_container_ids.add(value_id) + try: + return [ + _copy_pydantic_value( + item, + active_container_ids, + allow_models=allow_models, + declared_annotation=declared_annotation, + ) + for item in tuple.__iter__(value) + ] + finally: + active_container_ids.remove(value_id) + + raise TypeError("Unsupported value in tool approval payload") + + +def _merge_realized_declared_values( + explicit: Any, + realized: Any, + baseline: Any, +) -> Any: + """Keep realized declared values that differ from base-model defaults.""" + if type(explicit) is dict and type(realized) is dict and type(baseline) is dict: + merged = dict(explicit) + for key, realized_value in dict.items(realized): + if key not in baseline: + continue + baseline_value = baseline[key] + if key in explicit: + merged[key] = _merge_realized_declared_values( + explicit[key], + realized_value, + baseline_value, + ) + elif realized_value != baseline_value: + merged[key] = realized_value + return merged + if ( + type(explicit) is list + and type(realized) is list + and type(baseline) is list + and len(explicit) == len(realized) == len(baseline) + ): + return [ + _merge_realized_declared_values(explicit_item, realized_item, baseline_item) + for explicit_item, realized_item, baseline_item in zip( + explicit, + realized, + baseline, + strict=True, + ) + ] + return realized if realized != baseline else explicit + + +def _validate_declared_payload( + model_adapter: TypeAdapter[Any], + explicit: dict[str, Any], + realized: dict[str, Any], +) -> Any: + """Validate a declared payload after filling only missing required values.""" + while True: + try: + return model_adapter.validate_python(explicit) + except ValidationError as error: + filled_missing_value = False + for detail in error.errors( + include_url=False, + include_context=False, + include_input=False, + ): + if detail.get("type") != "missing": + continue + location = detail.get("loc") + if not isinstance(location, tuple) or not location: + continue + explicit_parent: Any = explicit + realized_parent: Any = realized + for part in location[:-1]: + if ( + type(part) is str + and type(explicit_parent) is dict + and type(realized_parent) is dict + and part in explicit_parent + and part in realized_parent + ): + explicit_parent = explicit_parent[part] + realized_parent = realized_parent[part] + elif ( + type(part) is int + and type(explicit_parent) is list + and type(realized_parent) is list + and 0 <= part < len(explicit_parent) + and part < len(realized_parent) + ): + explicit_parent = explicit_parent[part] + realized_parent = realized_parent[part] + else: + break + else: + missing_part = location[-1] + if ( + type(missing_part) is str + and type(explicit_parent) is dict + and type(realized_parent) is dict + and missing_part not in explicit_parent + and missing_part in realized_parent + ): + explicit_parent[missing_part] = realized_parent[missing_part] + filled_missing_value = True + if not filled_missing_value: + raise + + +def _restore_pydantic_fields_set(value: Any, source: Any) -> None: + """Restore declared field-set semantics after public Pydantic validation.""" + value_mro = _static_type_mro(value) + source_mro = _static_type_mro(source) + if BaseModel in value_mro and BaseModel in source_mro: + value_fields_set = BaseModel.model_fields_set.__get__(value, BaseModel) + source_fields_set = BaseModel.model_fields_set.__get__(source, BaseModel) + set.clear(value_fields_set) + set.update(value_fields_set, source_fields_set) + + source_values: dict[str, Any] = {} + for field_name, field_value in BaseModel.__iter__(source): + if str in _static_type_mro(field_name): + source_values[str.__str__(field_name)] = field_value + for field_name, field_value in BaseModel.__iter__(value): + if str not in _static_type_mro(field_name): + continue + source_value = source_values.get(str.__str__(field_name), _MISSING_CONTEXT_SENTINEL) + if source_value is not _MISSING_CONTEXT_SENTINEL: + _restore_pydantic_fields_set(field_value, source_value) + return + + if list in value_mro and list in source_mro: + for item, source_item in zip( + list.__iter__(value), + list.__iter__(source), + strict=False, + ): + _restore_pydantic_fields_set(item, source_item) + return + + if tuple in value_mro and tuple in source_mro: + for item, source_item in zip( + tuple.__iter__(value), + tuple.__iter__(source), + strict=False, + ): + _restore_pydantic_fields_set(item, source_item) + return + + if dict in value_mro and dict in source_mro: + for key, item in dict.items(value): + if str not in _static_type_mro(key): + continue + source_item = dict.get( + source, + str.__str__(key), + _MISSING_CONTEXT_SENTINEL, + ) + if source_item is not _MISSING_CONTEXT_SENTINEL: + _restore_pydantic_fields_set(item, source_item) + + +def _copy_tool_approval_raw_item(raw_item: Any) -> Any: + """Copy a supported approval raw item through public Pydantic APIs.""" + active_container_ids: set[int] = set() + raw_item_mro = _static_type_mro(raw_item) + for model_type, model_adapter in _TOOL_APPROVAL_MODEL_ADAPTERS: + if model_type not in raw_item_mro: + continue + copied_raw_item = _copy_pydantic_value( + raw_item, + active_container_ids, + allow_models=True, + declared_model_type=model_type, + ) + explicit = model_adapter.dump_python( + copied_raw_item, + mode="json", + round_trip=True, + exclude_unset=True, + warnings="error", + serialize_as_any=False, + by_alias=False, + ) + copied_explicit = _copy_json_compatible_value(explicit, active_container_ids) + if type(copied_explicit) is not dict: + raise TypeError("Unsupported serialized tool approval payload") + realized = model_adapter.dump_python( + copied_raw_item, + mode="json", + round_trip=True, + exclude_unset=False, + warnings="error", + serialize_as_any=False, + by_alias=False, + ) + copied_realized = _copy_json_compatible_value(realized, active_container_ids) + if type(copied_realized) is not dict: + raise TypeError("Unsupported serialized tool approval payload") + baseline_model = _validate_declared_payload( + model_adapter, + copied_explicit, + copied_realized, + ) + baseline = model_adapter.dump_python( + baseline_model, + mode="json", + round_trip=True, + exclude_unset=False, + warnings="error", + serialize_as_any=False, + by_alias=False, + ) + copied_baseline = _copy_json_compatible_value(baseline, active_container_ids) + merged = _merge_realized_declared_values( + copied_explicit, + copied_realized, + copied_baseline, + ) + validated_model = model_adapter.validate_python(merged) + _restore_pydantic_fields_set(validated_model, copied_raw_item) + return validated_model + if dict in raw_item_mro: + return _copy_json_compatible_value(raw_item, active_container_ids) + raise TypeError("Unsupported tool approval raw item") + + @dataclass class RunState(Generic[TContext, TAgent]): """Serializable snapshot of an agent run, including context, usage, and interruptions. @@ -452,20 +939,47 @@ def clear_pending_input(self) -> None: self._pending_input = [] def get_interruptions(self) -> list[ToolApprovalItem]: - """Return pending interruptions if the current step is an interruption.""" + """Return detached copies of pending interruptions for the current step.""" # Import at runtime to avoid circular import from .run_internal.run_steps import NextStepInterruption if self._current_step is None or not isinstance(self._current_step, NextStepInterruption): return [] - return list(self._current_step.interruptions) + copy_error: UserError | None = None + try: + interruptions: list[ToolApprovalItem] = [] + for item in self._current_step.interruptions: + copied_raw_item = _copy_tool_approval_raw_item(item.raw_item) + interruptions.append( + dataclasses.replace( + item, + agent=item.agent, + raw_item=copied_raw_item, + ) + ) + except Exception as error: + _prepare_data_redacted_error(error) + copy_error = UserError( + "Cannot safely copy pending tool approvals. Ensure each interruption uses a " + "supported tool call or contains only JSON-compatible mapping data." + ) + if copy_error is not None: + _mark_error_data_redacted(copy_error) + self = cast(Any, None) + item = cast(Any, None) + copied_raw_item = None + interruptions = [] + _raise_data_redacted_error(copy_error) + return interruptions @staticmethod def _approval_items_match( candidate: ToolApprovalItem, approval_item: ToolApprovalItem, - ) -> bool: - """Return whether two approval items identify the same nested invocation.""" + *, + approval_is_authoritative: bool = False, + ) -> bool | None: + """Compare approval identity, returning None when an owner is unsafe to distinguish.""" if candidate is approval_item: return True candidate_agent = candidate.agent @@ -476,18 +990,64 @@ def _approval_items_match( and candidate_agent is not approval_agent ): return False + try: + approval_raw_item = _copy_tool_approval_raw_item(approval_item.raw_item) + except Exception: + return None if approval_is_authoritative else False + try: + candidate_raw_item = _copy_tool_approval_raw_item(candidate.raw_item) + except Exception: + return None candidate_identity = tool_invocation_identity( - candidate.raw_item, + candidate_raw_item, tool_lookup_key=candidate.tool_lookup_key, tool_name=candidate.tool_name, ) approval_identity = tool_invocation_identity( - approval_item.raw_item, + approval_raw_item, tool_lookup_key=approval_item.tool_lookup_key, tool_name=approval_item.tool_name, ) return candidate_identity is not None and candidate_identity == approval_identity + def _find_current_approval_item( + self, + approval_item: ToolApprovalItem, + *, + approval_is_authoritative: bool | None = None, + ) -> ToolApprovalItem | None: + """Resolve a detached approval snapshot to current authoritative pending state.""" + from .run_internal.run_steps import NextStepInterruption + + if not isinstance(self._current_step, NextStepInterruption): + return None + if approval_is_authoritative is None: + approval_is_authoritative = any( + candidate is approval_item for candidate in self._current_step.interruptions + ) + canonical_matches: list[ToolApprovalItem] = [] + has_indeterminate_candidate = False + for candidate in self._current_step.interruptions: + if candidate is approval_item: + canonical_matches.append(candidate) + continue + match = self._approval_items_match( + candidate, + approval_item, + approval_is_authoritative=approval_is_authoritative, + ) + if match is None: + has_indeterminate_candidate = True + elif match: + canonical_matches.append(candidate) + if has_indeterminate_candidate or len(canonical_matches) > 1: + raise UserError( + "Cannot apply approval because multiple current pending approvals contain the " + "same tool invocation identity, or because it belongs to both the current run " + "and a nested agent-tool run. Use unique call IDs." + ) + return canonical_matches[0] if canonical_matches else None + def _find_nested_approval_state( self, approval_item: ToolApprovalItem, @@ -497,11 +1057,66 @@ def _find_nested_approval_state( return None from .agent_tool_state import peek_agent_tool_run_result + from .run_internal.run_steps import NextStepInterruption + nested_candidates: list[tuple[RunState[Any, Agent[Any]], ToolApprovalItem]] = [] + for function_run in self._last_processed_response.functions: + pending_result = peek_agent_tool_run_result( + function_run.tool_call, + scope_id=self._agent_tool_state_scope_id, + ) + interruptions = getattr(pending_result, "interruptions", None) + to_state = getattr(pending_result, "to_state", None) + if not isinstance(interruptions, list) or not callable(to_state): + continue + nested_state = to_state() + if not isinstance(nested_state, RunState) or nested_state is self: + continue + nested_candidates.extend( + (nested_state, candidate) + for candidate in interruptions + if isinstance(candidate, ToolApprovalItem) + ) + + current_candidates = ( + self._current_step.interruptions + if isinstance(self._current_step, NextStepInterruption) + else [] + ) + approval_is_authoritative = any( + candidate is approval_item for candidate in current_candidates + ) or any(candidate is approval_item for _, candidate in nested_candidates) + current_approval_item = self._find_current_approval_item( + approval_item, + approval_is_authoritative=approval_is_authoritative, + ) + canonical_matches: list[tuple[RunState[Any, Agent[Any]], ToolApprovalItem]] = [] + has_indeterminate_candidate = False + for nested_state, candidate in nested_candidates: + if candidate is approval_item: + canonical_matches.append((nested_state, candidate)) + continue + match = self._approval_items_match( + candidate, + approval_item, + approval_is_authoritative=approval_is_authoritative, + ) + if match is None: + has_indeterminate_candidate = True + elif match: + canonical_matches.append((nested_state, candidate)) + + if has_indeterminate_candidate: + raise UserError( + "Cannot apply approval because one or more nested agent-tool approvals cannot be " + "safely distinguished. Use JSON-compatible approval payloads and unique call IDs." + ) + + identity_item = current_approval_item or approval_item approval_identity = tool_invocation_identity_and_scope( - approval_item.raw_item, - tool_lookup_key=approval_item.tool_lookup_key, - tool_name=approval_item.tool_name, + identity_item.raw_item, + tool_lookup_key=identity_item.tool_lookup_key, + tool_name=identity_item.tool_name, ) current_state_owns_approval = False if approval_identity is not None and self._context is not None: @@ -579,38 +1194,11 @@ def _find_nested_approval_state( current_state_owns_approval and approval_identity in current_response_identities ) - exact_match: tuple[RunState[Any, Agent[Any]], ToolApprovalItem] | None = None - canonical_matches: list[tuple[RunState[Any, Agent[Any]], ToolApprovalItem]] = [] - for function_run in self._last_processed_response.functions: - pending_result = peek_agent_tool_run_result( - function_run.tool_call, - scope_id=self._agent_tool_state_scope_id, - ) - interruptions = getattr(pending_result, "interruptions", None) - to_state = getattr(pending_result, "to_state", None) - if not isinstance(interruptions, list) or not callable(to_state): - continue - nested_state = to_state() - if not isinstance(nested_state, RunState) or nested_state is self: - continue - for candidate in interruptions: - if not isinstance(candidate, ToolApprovalItem): - continue - if candidate is approval_item: - exact_match = (nested_state, candidate) - break - if self._approval_items_match(candidate, approval_item): - canonical_matches.append((nested_state, candidate)) - if exact_match is not None: - break - - if current_state_owns_approval and (exact_match is not None or canonical_matches): + if current_state_owns_approval and canonical_matches: raise UserError( "Cannot apply approval because the same tool invocation identity belongs to both " "the current run and a nested agent-tool run. Use distinct call IDs." ) - if exact_match is not None: - return exact_match if len(canonical_matches) == 1: return canonical_matches[0] if len(canonical_matches) > 1: @@ -629,7 +1217,11 @@ def approve(self, approval_item: ToolApprovalItem, always_approve: bool = False) nested_state, nested_item = nested_approval nested_state.approve(nested_item, always_approve=always_approve) return - self._context.approve_tool(approval_item, always_approve=always_approve) + current_approval_item = self._find_current_approval_item(approval_item) + self._context.approve_tool( + current_approval_item or approval_item, + always_approve=always_approve, + ) def reject( self, @@ -656,7 +1248,7 @@ def reject( ) return self._context.reject_tool( - approval_item, + self._find_current_approval_item(approval_item) or approval_item, always_reject=always_reject, rejection_message=rejection_message, ) diff --git a/tests/test_run_state.py b/tests/test_run_state.py index 748d0c064e..21475ca326 100644 --- a/tests/test_run_state.py +++ b/tests/test_run_state.py @@ -8,14 +8,16 @@ import logging from collections.abc import Callable, Mapping from copy import deepcopy -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import datetime from pathlib import Path from types import SimpleNamespace -from typing import Any, TypeVar, cast +from typing import Any, ClassVar, Literal, TypeVar, cast import pytest from openai.types.responses import ( + ResponseCustomToolCall, + ResponseFunctionShellToolCall, ResponseFunctionToolCall, ResponseOutputMessage, ResponseOutputText, @@ -30,13 +32,15 @@ from openai.types.responses.response_function_tool_call import CallerProgram from openai.types.responses.response_output_item import ( LocalShellCall, + LocalShellCallAction, McpApprovalRequest, + McpCall, Program, ProgramOutput, ) from openai.types.responses.response_usage import InputTokensDetails from openai.types.responses.tool_param import Mcp -from pydantic import BaseModel, ValidationError +from pydantic import BaseModel, ValidationError, model_serializer from agents import Agent, ModelSettings, RunConfig, RunHooks, Runner, handoff, trace from agents._tool_invocation import tool_invocation_identity_and_scope @@ -1612,26 +1616,1318 @@ def test_get_interruptions_returns_interruptions_when_present(self): interruptions = state.get_interruptions() assert len(interruptions) == 1 - assert interruptions[0] == approval_item + assert interruptions[0] is not approval_item + assert interruptions[0].agent is agent + assert interruptions[0].tool_name == approval_item.tool_name + assert interruptions[0].raw_item.model_dump() == approval_item.raw_item.model_dump() + assert interruptions[0].raw_item is not approval_item.raw_item + + @pytest.mark.parametrize("raw_item_kind", ["pydantic", "mapping"]) + def test_get_interruptions_returns_detached_item_snapshots(self, raw_item_kind: str): + """Mutating returned interruption content must not change pending approvals.""" + agent = Agent(name="SnapshotAgent") + raw_item: Any + if raw_item_kind == "pydantic": + raw_item = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="cid-snapshot", + status="completed", + arguments='{"value": "original"}', + ) + else: + raw_item = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-snapshot", + "status": "completed", + "arguments": '{"value": "original"}', + "metadata": {"tags": ["original"]}, + } + approval_item = ToolApprovalItem( + agent=agent, + raw_item=raw_item, + ) + state = make_state_with_interruptions(agent, [approval_item]) + + interruption = state.get_interruptions()[0] + interruption.tool_name = "changed" + if isinstance(interruption.raw_item, dict): + interruption.raw_item["arguments"] = '{"value": "changed"}' + interruption.raw_item["metadata"]["tags"].append("changed") + else: + interruption.raw_item.arguments = '{"value": "changed"}' + + pending = state.get_interruptions()[0] + assert pending is not interruption + assert pending.agent is agent + assert pending.tool_name == "toolA" + if isinstance(pending.raw_item, dict): + assert pending.raw_item["arguments"] == '{"value": "original"}' + assert pending.raw_item["metadata"] == {"tags": ["original"]} + else: + assert pending.raw_item.arguments == '{"value": "original"}' + + @pytest.mark.parametrize("raw_item_kind", ["pydantic", "mapping"]) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_get_interruptions_snapshots_can_apply_approval_decisions( + self, + raw_item_kind: str, + approve: bool, + ) -> None: + """Detached snapshots must retain canonical approval identity.""" + agent = Agent(name="DecisionAgent") + raw_item: Any = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-decision", + "status": "completed", + "arguments": "{}", + } + if raw_item_kind == "pydantic": + raw_item = ResponseFunctionToolCall(**raw_item) + approval_item = ToolApprovalItem(agent=agent, raw_item=raw_item) + state = make_state_with_interruptions(agent, [approval_item]) + + interruption = state.get_interruptions()[0] + assert interruption is not approval_item + if approve: + state.approve(interruption) + else: + state.reject(interruption) + + assert state._context is not None + assert state._context.is_tool_approved("toolA", "cid-decision") is approve + + def test_get_interruptions_fails_before_returning_an_unsafe_snapshot(self): + """Uncopyable payloads must fail at the snapshot boundary.""" + + class Uncopyable: + def __deepcopy__(self, _memo: dict[int, Any]) -> Any: + raise RuntimeError("cannot copy") + + agent = Agent(name="UncopyableAgent") + approval_item = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "toolA", + "call_id": "cid-uncopyable", + "status": "completed", + "arguments": "{}", + "metadata": Uncopyable(), + }, + ) + state = make_state_with_interruptions(agent, [approval_item]) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + def test_get_interruptions_clone_failure_drops_sensitive_exception_context(self) -> None: + """Clone failures must not retain payload data in the exception graph.""" + source_sentinel = "SENSITIVE_APPROVAL_CONTEXT_VALUE" + partial_sentinel = "SENSITIVE_PARTIAL_COPY_VALUE" + agent = Agent(name="CloneFailureContextAgent") + safe_item = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "safeTool", + "call_id": "cid-safe-before-sensitive-failure", + "arguments": "{}", + "metadata": {"secret": partial_sentinel}, + }, + ) + raw_item = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-sensitive-clone-failure", + "arguments": "{}", + "metadata": {"secret": source_sentinel, "unsafe": object()}, + } + failing_item = ToolApprovalItem(agent=agent, raw_item=raw_item) + state = make_state_with_interruptions( + agent, + [safe_item, failing_item], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals") as exc: + state.get_interruptions() + + assert exc.value.__cause__ is None + assert exc.value.__context__ is None + assert source_sentinel not in repr(exc.value) + assert partial_sentinel not in repr(exc.value) + traceback = exc.value.__traceback__ + while traceback is not None: + frame = traceback.tb_frame + if "/src/agents/" in frame.f_code.co_filename: + local_values = tuple(frame.f_locals.values()) + assert all(value is not state for value in local_values) + assert all(value is not safe_item for value in local_values) + assert all(value is not failing_item for value in local_values) + assert all(value is not raw_item for value in local_values) + assert not any(isinstance(value, RunState) for value in local_values) + assert not any(isinstance(value, ToolApprovalItem) for value in local_values) + assert source_sentinel not in repr(frame.f_locals) + assert partial_sentinel not in repr(frame.f_locals) + traceback = traceback.tb_next + + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_get_interruptions_canonicalizes_custom_outer_models( + self, + approve: bool, + ) -> None: + """Declared model subtypes must retain canonical approval identity.""" + + class CustomCall(ResponseFunctionToolCall): + serializer_called: ClassVar[bool] = False + status: Literal["completed"] = "completed" + action: dict[str, str] + subtype_metadata: dict[str, list[str]] + subtype_only: Any + + @model_serializer(mode="wrap") + def serialize_custom_call(self, handler: Any) -> Any: + type(self).serializer_called = True + return handler(self) + + raw_item = CustomCall( + type="function_call", + name="toolA", + call_id="cid-custom-model", + arguments="{}", + action={"kind": "subtype-only"}, + subtype_metadata={"tags": ["subtype-only"]}, + subtype_only=object(), + ) + agent = Agent(name="CustomModelAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + assert state._context is not None + state._context._tool_invocation_status(raw_item) + CustomCall.serializer_called = False + snapshot = state.get_interruptions()[0] + + assert not CustomCall.serializer_called + assert type(snapshot.raw_item) is ResponseFunctionToolCall + assert snapshot.raw_item.call_id == "cid-custom-model" + assert snapshot.raw_item.status == "completed" + assert "status" not in snapshot.raw_item.model_fields_set + assert "status" not in snapshot.raw_item.model_dump(exclude_unset=True) + assert "subtype_metadata" not in snapshot.raw_item.model_dump() + if approve: + state.approve(snapshot) + else: + state.reject(snapshot) + assert state._context.is_tool_approved("toolA", "cid-custom-model") is approve + + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_get_interruptions_subtype_snapshots_route_to_nested_approval( + self, + approve: bool, + ) -> None: + """Canonical subtype snapshots must resolve to nested authoritative items.""" + from agents.agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result + + class CustomCall(ResponseFunctionToolCall): + status: Literal["completed"] = "completed" + action: dict[str, str] + + agent = Agent(name="NestedSubtypeAgent") + nested_tool = function_tool(lambda: "nested", name_override="nested_agent_tool") + nested_outer_call = make_tool_call( + call_id="outer-nested-subtype", + name="nested_agent_tool", + ) + raw_item = CustomCall( + type="function_call", + name="toolA", + call_id="cid-nested-subtype", + arguments="{}", + action={"kind": "subtype-only"}, + ) + nested_approval = ToolApprovalItem(agent=agent, raw_item=raw_item) + state = make_state_with_interruptions(agent, [nested_approval]) + state._last_processed_response = make_processed_response( + functions=[ + ToolRunFunction(tool_call=nested_outer_call, function_tool=nested_tool), + ] + ) + nested_state = make_state_with_interruptions(agent, [nested_approval]) + assert nested_state._context is not None + nested_state._context._tool_invocation_status(raw_item) + record_agent_tool_run_result( + nested_outer_call, + cast( + Any, + SimpleNamespace( + interruptions=[nested_approval], + to_state=lambda: nested_state, + ), + ), + scope_id=state._agent_tool_state_scope_id, + ) + + try: + snapshot = state.get_interruptions()[0] + assert snapshot.raw_item.status == "completed" + if approve: + state.approve(snapshot) + else: + state.reject(snapshot) + assert ( + nested_state._context.is_tool_approved( + "toolA", + "cid-nested-subtype", + ) + is approve + ) + finally: + drop_agent_tool_run_result( + nested_outer_call, + scope_id=state._agent_tool_state_scope_id, + ) + + def test_get_interruptions_preserves_required_declared_subtype_defaults(self) -> None: + """Subtype defaults for base-required fields must survive canonicalization.""" + + class DefaultArgumentsCall(ResponseFunctionToolCall): + arguments: str = "{}" + + raw_item = DefaultArgumentsCall( + type="function_call", + name="toolA", + call_id="cid-required-subtype-default", + ) + agent = Agent(name="RequiredSubtypeDefaultAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + snapshot = state.get_interruptions()[0] + + assert type(snapshot.raw_item) is ResponseFunctionToolCall + assert snapshot.raw_item.arguments == "{}" + assert "arguments" not in snapshot.raw_item.model_fields_set + assert "arguments" not in snapshot.raw_item.model_dump(exclude_unset=True) + assert raw_item.arguments == "{}" + + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_get_interruptions_rejects_typed_extra_identity_collisions( + self, + approve: bool, + ) -> None: + """Typed extras must not replace declared approval identity fields.""" + agent = Agent(name="TypedExtraCollisionAgent") + raw_item = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="authoritative", + arguments="{}", + ) + assert raw_item.model_extra is not None + raw_item.model_extra["call_id"] = "forged" + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + snapshot = state.get_interruptions()[0] + if approve: + state.approve(snapshot) + else: + state.reject(snapshot) + + assert state._context is not None + assert state._context.is_tool_approved("toolA", "authoritative") is None + assert state._context.is_tool_approved("toolA", "forged") is None + + def test_get_interruptions_does_not_hash_typed_extra_keys(self) -> None: + """Typed-extra keys must be normalized before any hash-based lookup.""" + + class MutatingKey(str): + def __new__(cls, value: str, owner: ResponseFunctionToolCall) -> MutatingKey: + key = str.__new__(cls, value) + key.owner = owner + return key + + def __hash__(self) -> int: + object.__setattr__(self.owner, "arguments", "mutated-by-key-hash") + return str.__hash__(self) + + agent = Agent(name="TypedExtraKeyAgent") + raw_item = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="cid-typed-extra-key", + arguments="original", + ) + assert raw_item.model_extra is not None + key = MutatingKey("metadata", raw_item) + raw_item.model_extra[key] = {"safe": True} + object.__setattr__(raw_item, "arguments", "original") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + snapshot = state.get_interruptions()[0] + + assert raw_item.arguments == "original" + assert snapshot.raw_item.arguments == "original" + snapshot_extra = snapshot.raw_item.model_extra + assert snapshot_extra == {"metadata": {"safe": True}} + assert snapshot_extra is not None + assert all(type(extra_name) is str for extra_name in snapshot_extra) + + def test_get_interruptions_copies_typed_extra_container_subtypes(self) -> None: + """Hook-free built-in container subtypes remain detached and supported.""" + + class PlainDict(dict[str, Any]): + pass + + class PlainList(list[str]): + pass + + metadata = PlainDict(tags=PlainList(["original"])) + raw_item = ResponseFunctionToolCall.model_validate( + { + "type": "function_call", + "name": "toolA", + "call_id": "cid-typed-extra-containers", + "arguments": "{}", + "metadata": metadata, + } + ) + agent = Agent(name="TypedExtraContainerAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + snapshot = state.get_interruptions()[0] + + assert raw_item.model_extra is not None + assert snapshot.raw_item.model_extra is not None + source_metadata = raw_item.model_extra["metadata"] + copied_metadata = snapshot.raw_item.model_extra["metadata"] + assert isinstance(source_metadata, PlainDict) + assert isinstance(source_metadata["tags"], PlainList) + assert type(copied_metadata) is dict + assert type(copied_metadata["tags"]) is list + copied_metadata["tags"].append("changed") + assert source_metadata["tags"] == ["original"] + + @pytest.mark.parametrize("location", ["outer", "nested"]) + def test_get_interruptions_rejects_serializer_bearing_typed_extras( + self, + location: str, + ) -> None: + """Typed extras must fail before a user serializer can mutate pending state.""" + + class MutatingExtra(BaseModel): + serializer_called: ClassVar[bool] = False + value: str + + @model_serializer(mode="wrap") + def serialize_mutating_extra(self, handler: Any) -> Any: + type(self).serializer_called = True + self.value = "mutated-by-serializer" + return handler(self) + + extra = MutatingExtra(value="original") + if location == "outer": + raw_item: Any = ResponseFunctionToolCall.model_validate( + { + "type": "function_call", + "name": "toolA", + "call_id": "cid-serializer-extra", + "arguments": "{}", + "metadata": extra, + } + ) + else: + raw_item = LocalShellCall.model_validate( + { + "id": "local-shell-serializer-extra", + "action": LocalShellCallAction.model_validate( + { + "command": ["echo", "ok"], + "env": {}, + "type": "exec", + "metadata": extra, + } + ), + "call_id": "cid-serializer-extra", + "status": "completed", + "type": "local_shell_call", + } + ) + agent = Agent(name="SerializerExtraAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + assert extra.value == "original" + assert not MutatingExtra.serializer_called + + @pytest.mark.parametrize("location", ["outer", "nested"]) + def test_get_interruptions_rejects_typed_subtype_attribute_hooks( + self, + location: str, + ) -> None: + """Subtype attribute hooks must fail before authoritative model access.""" + + class MutatingCall(ResponseFunctionToolCall): + armed: bool = False + + def __getattribute__(self, name: str) -> Any: + if name in {"__dict__", "__pydantic_extra__"} and object.__getattribute__( + self, + "__dict__", + ).get("armed"): + object.__setattr__(self, "arguments", '{"mutated":true}') + return super().__getattribute__(name) + + class MutatingAction(LocalShellCallAction): + armed: bool = False + + def __getattribute__(self, name: str) -> Any: + if name in {"__dict__", "__pydantic_extra__"} and object.__getattribute__( + self, + "__dict__", + ).get("armed"): + object.__setattr__(self, "command", ["mutated"]) + return super().__getattribute__(name) + + if location == "outer": + raw_item: Any = MutatingCall( + type="function_call", + name="toolA", + call_id="cid-hook-bearing-subtype", + arguments="{}", + ) + raw_item.armed = True + else: + action = MutatingAction(command=["echo", "ok"], env={}, type="exec") + action.armed = True + raw_item = LocalShellCall( + id="local-shell-hook-bearing-subtype", + action=action, + call_id="cid-hook-bearing-subtype", + status="completed", + type="local_shell_call", + ) + agent = Agent(name="HookBearingSubtypeAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + if location == "outer": + assert raw_item.arguments == "{}" + else: + assert raw_item.action.command == ["echo", "ok"] + + def test_get_interruptions_rejects_hooks_in_post_declared_model_mixins(self) -> None: + """Subtype hooks must be rejected even when their mixin follows the declared base.""" + + class MutatingMixin: + def __getattribute__(self, name: str) -> Any: + if name in {"__dict__", "__pydantic_extra__"} and object.__getattribute__( + self, + "__dict__", + ).get("armed"): + object.__setattr__(self, "arguments", "mutated-by-post-declared-mixin") + return super().__getattribute__(name) + + class MutatingCall(ResponseFunctionToolCall, MutatingMixin): + armed: bool = False + + raw_item = MutatingCall( + type="function_call", + name="toolA", + call_id="cid-post-declared-mixin", + arguments="original", + ) + agent = Agent(name="PostDeclaredMixinAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + raw_item.armed = True + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + assert raw_item.arguments == "original" + + @pytest.mark.parametrize( + ("location", "storage_name"), + [ + ("outer", "__pydantic_extra__"), + ("nested", "__pydantic_fields_set__"), + ("outer", "__dict__"), + ], + ) + def test_get_interruptions_rejects_pydantic_storage_descriptors( + self, + location: str, + storage_name: str, + ) -> None: + """Subtype storage descriptors must fail before public Pydantic instance access.""" + hook_called = False + source_holder: dict[str, Any] = {} + + def mutate_on_access(_instance: BaseModel) -> Any: + nonlocal hook_called + hook_called = True + source_model = source_holder["model"] + object.__setattr__( + source_model, + source_holder["field"], + source_holder["mutated_value"], + ) + raise AssertionError("storage descriptor should not run") + + class CustomCall(ResponseFunctionToolCall): + pass + + class DictDescriptorCall(ResponseFunctionToolCall): + __dict__ = property(mutate_on_access) # type: ignore[assignment] + + class CustomAction(LocalShellCallAction): + pass + + if location == "outer": + call_type = ResponseFunctionToolCall if storage_name == "__dict__" else CustomCall + raw_item: Any = call_type( + type="function_call", + name="toolA", + call_id="cid-storage-descriptor", + arguments="original", + ) + if storage_name == "__dict__": + object.__setattr__(raw_item, "__class__", DictDescriptorCall) + source_model = raw_item + source_field = "arguments" + mutated_value: Any = "mutated-by-storage-descriptor" + else: + source_model = CustomAction(command=["echo", "ok"], env={}, type="exec") + raw_item = LocalShellCall( + id="local-shell-storage-descriptor", + action=source_model, + call_id="cid-storage-descriptor", + status="completed", + type="local_shell_call", + ) + source_field = "command" + mutated_value = ["mutated-by-storage-descriptor"] + + source_holder.update( + model=source_model, + field=source_field, + mutated_value=mutated_value, + ) + if storage_name != "__dict__": + setattr(type(source_model), storage_name, property(mutate_on_access)) + agent = Agent(name="StorageDescriptorAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + assert not hook_called + if location == "outer": + assert source_model.arguments == "original" + else: + assert source_model.command == ["echo", "ok"] + + @pytest.mark.parametrize("location", ["outer", "nested"]) + def test_get_interruptions_rejects_pydantic_storage_container_hooks( + self, + location: str, + ) -> None: + """Pydantic storage containers must be plain dicts before public iteration.""" + + class MutatingStorage(dict[str, Any]): + def __init__(self, *args: Any, field: str, mutated_value: Any) -> None: + super().__init__(*args) + self.field = field + self.mutated_value = mutated_value + + def items(self) -> Any: + self[self.field] = self.mutated_value + return super().items() + + if location == "outer": + raw_item: Any = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="cid-storage-container", + arguments="original", + ) + source_model = raw_item + source_field = "arguments" + mutated_value: Any = "mutated-by-storage-container" + else: + source_model = LocalShellCallAction(command=["echo", "ok"], env={}, type="exec") + raw_item = LocalShellCall( + id="local-shell-storage-container", + action=source_model, + call_id="cid-storage-container", + status="completed", + type="local_shell_call", + ) + source_field = "command" + mutated_value = ["mutated-by-storage-container"] + + storage = MutatingStorage( + object.__getattribute__(source_model, "__dict__"), + field=source_field, + mutated_value=mutated_value, + ) + object.__setattr__(source_model, "__dict__", storage) + agent = Agent(name="StorageContainerAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + if location == "outer": + assert source_model.arguments == "original" + else: + assert source_model.command == ["echo", "ok"] + + @pytest.mark.parametrize("location", ["outer", "nested"]) + def test_get_interruptions_does_not_dispatch_pydantic_storage_key_hooks( + self, + location: str, + ) -> None: + """Pydantic storage keys must be normalized without method dispatch.""" + hook_called = False + + class MutatingKey(str): + def __new__( + cls, + value: str, + owner: BaseModel, + field: str, + mutated_value: Any, + ) -> MutatingKey: + key = str.__new__(cls, value) + key.owner = owner + key.field = field + key.mutated_value = mutated_value + return key + + def startswith(self, *args: Any, **kwargs: Any) -> bool: + nonlocal hook_called + hook_called = True + object.__setattr__(self.owner, self.field, self.mutated_value) + return str.startswith(self, *args, **kwargs) + + if location == "outer": + raw_item: Any = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="cid-storage-key", + arguments="original", + ) + source_model = raw_item + source_field = "arguments" + original_value: Any = "original" + mutated_value: Any = "mutated-by-storage-key" + else: + source_model = LocalShellCallAction(command=["echo", "ok"], env={}, type="exec") + raw_item = LocalShellCall( + id="local-shell-storage-key", + action=source_model, + call_id="cid-storage-key", + status="completed", + type="local_shell_call", + ) + source_field = "command" + original_value = ["echo", "ok"] + mutated_value = ["mutated-by-storage-key"] + + storage = object.__getattribute__(source_model, "__dict__") + assert type(storage) is dict + dict.__setitem__( + storage, + MutatingKey( + "subtype_only", + source_model, + source_field, + mutated_value, + ), + "ignored", + ) + agent = Agent(name="StorageKeyAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + snapshot = state.get_interruptions()[0] + + assert not hook_called + assert getattr(source_model, source_field) == original_value + snapshot_raw_item = cast(Any, snapshot.raw_item) + snapshot_model = snapshot_raw_item if location == "outer" else snapshot_raw_item.action + assert getattr(snapshot_model, source_field) == original_value + + @pytest.mark.parametrize("location", ["mapping", "typed_extra"]) + def test_get_interruptions_does_not_dispatch_payload_class_properties( + self, + location: str, + ) -> None: + """Classifying arbitrary payload values must not access their __class__.""" + + class MutatingClassProbe: + def __init__(self, mutate: Callable[[], None]) -> None: + self.mutate = mutate + + @property + def __class__(self) -> type[object]: + self.mutate() + return object + + if location == "mapping": + raw_item: Any = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-class-property", + "arguments": "original", + } + probe = MutatingClassProbe( + lambda: raw_item.__setitem__("arguments", "mutated-by-class-property") + ) + raw_item["metadata"] = probe + else: + raw_item = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="cid-class-property", + arguments="original", + ) + probe = MutatingClassProbe( + lambda: object.__setattr__( + raw_item, + "arguments", + "mutated-by-class-property", + ) + ) + assert raw_item.model_extra is not None + raw_item.model_extra["metadata"] = probe + agent = Agent(name="ClassPropertyAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + if location == "mapping": + assert raw_item["arguments"] == "original" + else: + assert raw_item.arguments == "original" + + def test_get_interruptions_checks_later_adapter_subtypes_without_class_access(self) -> None: + """Adapter selection must reject hooks without reading instance __class__.""" + + class MutatingMcpCall(McpCall): + armed: bool = False + + def __getattribute__(self, name: str) -> Any: + if name == "__class__" and object.__getattribute__(self, "__dict__").get("armed"): + object.__setattr__(self, "arguments", "mutated-by-adapter-selection") + return super().__getattribute__(name) + + raw_item = MutatingMcpCall( + id="mcp-hook-bearing-subtype", + arguments="original", + name="toolA", + server_label="server", + type="mcp_call", + ) + agent = Agent(name="LaterAdapterSubtypeAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + raw_item.armed = True + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + assert raw_item.arguments == "original" + + def test_get_interruptions_uses_public_schema_for_nested_models(self) -> None: + """Base adapters must serialize nested models without subclass serializers.""" + + class CustomAction(LocalShellCallAction): + serializer_called: ClassVar[bool] = False + command: list[str] = ["echo", "ok"] + subtype_only: Any + + @model_serializer(mode="wrap") + def serialize_custom_action(self, handler: Any) -> Any: + type(self).serializer_called = True + return handler(self) + + action = CustomAction( + env={}, + type="exec", + subtype_only=object(), + ) + raw_item = LocalShellCall( + id="local-shell-public-schema", + action=action, + call_id="cid-local-shell-public-schema", + status="completed", + type="local_shell_call", + ) + agent = Agent(name="PublicSchemaAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + snapshot = state.get_interruptions()[0] + + assert type(snapshot.raw_item) is LocalShellCall + assert snapshot.raw_item.action.command == ["echo", "ok"] + assert type(snapshot.raw_item.action) is LocalShellCallAction + assert snapshot.raw_item.action is not action + assert "command" not in snapshot.raw_item.action.model_fields_set + assert "command" not in snapshot.raw_item.action.model_dump(exclude_unset=True) + assert not CustomAction.serializer_called + + def test_get_interruptions_uses_trusted_nested_model_annotations(self) -> None: + """Nested model discovery must not trust a caller-controlled module name.""" + source_holder: dict[str, Any] = {} + + class SpoofedAction(LocalShellCallAction): + construct_called: ClassVar[bool] = False + + @classmethod + def model_construct( + cls, + _fields_set: set[str] | None = None, + **values: object, + ) -> Any: + cls.construct_called = True + source_holder["action"].command = ["mutated"] + return super().model_construct(_fields_set=_fields_set, **values) + + SpoofedAction.__module__ = "openai.types.responses.spoofed" + action = SpoofedAction(command=["echo", "ok"], env={}, type="exec") + source_holder["action"] = action + raw_item = LocalShellCall( + id="local-shell-spoofed-module", + action=action, + call_id="cid-spoofed-module", + status="completed", + type="local_shell_call", + ) + agent = Agent(name="SpoofedNestedModelAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + snapshot = state.get_interruptions()[0] + + assert not SpoofedAction.construct_called + assert action.command == ["echo", "ok"] + assert type(snapshot.raw_item.action) is LocalShellCallAction + assert snapshot.raw_item.action.command == ["echo", "ok"] + + @pytest.mark.parametrize("location", ["mapping", "nested"]) + def test_get_interruptions_rejects_normalized_key_collisions(self, location: str) -> None: + """Distinct source keys must not collapse into one approval identity field.""" + + class DistinctString(str): + def __hash__(self) -> int: + return id(self) + + def __eq__(self, other: object) -> bool: + return self is other + + colliding_key = DistinctString("call_id") + agent = Agent(name="KeyCollisionAgent") + if location == "mapping": + raw_item: Any = { + "type": "function_call", + "name": "toolA", + "call_id": "original", + "arguments": "{}", + colliding_key: "replacement", + } + else: + metadata = {"call_id": "original", colliding_key: "replacement"} + raw_item = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-nested-collision", + "arguments": "{}", + "metadata": metadata, + } + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + if location == "nested": + assert metadata["call_id"] == "original" + assert metadata[colliding_key] == "replacement" + else: + assert raw_item["call_id"] == "original" + + def test_get_interruptions_bypasses_nested_container_hooks(self): + """Mapping snapshots must not invoke hooks on container subclasses.""" + + class MutatingDict(dict[str, Any]): + def items(self) -> Any: + self["serializer-side-effect"] = True + return super().items() + + class MutatingList(list[str]): + def __iter__(self) -> Any: + self.append("serializer-side-effect") + return super().__iter__() + + metadata = MutatingList(["original"]) + raw_item = MutatingDict( + type="function_call", + name="toolA", + call_id="cid-hooks", + arguments="{}", + metadata=metadata, + ) + agent = Agent(name="ContainerHooksAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + interruption = state.get_interruptions()[0] + + assert type(interruption.raw_item) is dict + assert interruption.raw_item["metadata"] == ["original"] + assert dict.__contains__(raw_item, "serializer-side-effect") is False + assert list.__len__(metadata) == 1 + + @pytest.mark.parametrize("non_finite", [float("nan"), float("inf"), float("-inf")]) + def test_get_interruptions_rejects_non_finite_mapping_values( + self, + non_finite: float, + ) -> None: + """Non-standard JSON numbers must fail before a snapshot is returned.""" + agent = Agent(name="NonFiniteAgent") + raw_item = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-non-finite", + "arguments": "{}", + "metadata": non_finite, + } + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + def test_get_interruptions_failure_does_not_expose_partial_snapshots(self): + """A later unsafe payload must fail without changing earlier pending items.""" + agent = Agent(name="PartialFailureAgent") + first_raw_item = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-safe", + "arguments": "original", + } + second_raw_item = { + "type": "function_call", + "name": "toolB", + "call_id": "cid-unsafe", + "arguments": "{}", + "metadata": object(), + } + state = make_state_with_interruptions( + agent, + [ + ToolApprovalItem(agent=agent, raw_item=first_raw_item), + ToolApprovalItem(agent=agent, raw_item=second_raw_item), + ], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + assert first_raw_item["arguments"] == "original" + + def test_get_interruptions_rejects_unsafe_typed_extra_metadata( + self, + ) -> None: + """Unsafe typed extras must fail through the same bounded copy path.""" + agent = Agent(name="TypedExtraFailureAgent") + raw_item = ResponseFunctionToolCall.model_validate( + { + "type": "function_call", + "name": "toolA", + "call_id": "cid-typed-extra", + "arguments": "{}", + "metadata": {"nested": object()}, + } + ) + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + def test_get_interruptions_rejects_cyclic_mapping_data(self) -> None: + """Cyclic mapping content must fail without changing authoritative state.""" + metadata: list[Any] = [] + metadata.append(metadata) + agent = Agent(name="CyclicMappingAgent") + raw_item = { + "type": "function_call", + "name": "toolA", + "call_id": "cid-cycle", + "arguments": "{}", + "metadata": metadata, + } + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + with pytest.raises(UserError, match="Cannot safely copy pending tool approvals"): + state.get_interruptions() + + assert len(metadata) == 1 + assert metadata[0] is metadata + + @pytest.mark.parametrize( + "raw_item", + [ + ResponseFunctionToolCall.model_validate( + { + "type": "function_call", + "name": "toolA", + "call_id": "cid-function", + "status": "completed", + "arguments": "{}", + "metadata": {"tags": ["function"]}, + } + ), + ResponseCustomToolCall.model_validate( + { + "type": "custom_tool_call", + "name": "toolA", + "call_id": "cid-custom", + "input": "original", + "metadata": {"tags": ["custom"]}, + } + ), + ResponseFunctionShellToolCall.model_validate( + { + "id": "shell-call", + "action": { + "commands": ["echo", "ok"], + "metadata": {"tags": ["action"]}, + }, + "call_id": "cid-shell", + "status": "completed", + "type": "shell_call", + "metadata": {"tags": ["shell"]}, + } + ), + McpCall.model_validate( + { + "id": "mcp-call", + "arguments": "{}", + "name": "toolA", + "server_label": "server", + "type": "mcp_call", + "metadata": {"tags": ["mcp-call"]}, + } + ), + McpApprovalRequest.model_validate( + { + "id": "mcp-approval", + "arguments": "{}", + "name": "toolA", + "server_label": "server", + "type": "mcp_approval_request", + "metadata": {"tags": ["mcp-approval"]}, + } + ), + LocalShellCall.model_validate( + { + "id": "local-shell", + "action": LocalShellCallAction.model_validate( + { + "command": ["echo", "ok"], + "env": {}, + "type": "exec", + "metadata": {"tags": ["action"]}, + } + ), + "call_id": "cid-local-shell", + "status": "completed", + "type": "local_shell_call", + "metadata": {"tags": ["local-shell"]}, + } + ), + ], + ids=["function", "custom", "shell", "mcp-call", "mcp-approval", "local-shell"], + ) + def test_get_interruptions_copies_each_typed_raw_item(self, raw_item: Any) -> None: + """Each declared typed approval payload must be reconstructed safely.""" + agent = Agent(name="TypedRawItemAgent") + state = make_state_with_interruptions( + agent, + [ToolApprovalItem(agent=agent, raw_item=raw_item)], + ) + + interruption = state.get_interruptions()[0] + + assert isinstance(interruption.raw_item, type(raw_item)) + assert interruption.raw_item.model_dump() == raw_item.model_dump() + assert interruption.raw_item is not raw_item + assert interruption.raw_item.model_fields_set == raw_item.model_fields_set + interruption_extra = interruption.raw_item.model_extra + raw_extra = raw_item.model_extra + assert interruption_extra == raw_extra + assert interruption_extra is not None + assert raw_extra is not None + assert interruption_extra is not raw_extra + + interruption_extra["metadata"]["tags"].append("changed") + assert raw_extra["metadata"]["tags"][-1] != "changed" + if isinstance(raw_item, LocalShellCall | ResponseFunctionShellToolCall): + interruption_action_extra = interruption.raw_item.action.model_extra + raw_action_extra = raw_item.action.model_extra + assert interruption_action_extra == raw_action_extra + assert interruption_action_extra is not None + assert raw_action_extra is not None + assert interruption_action_extra is not raw_action_extra + interruption_action_extra["metadata"]["tags"].append("changed") + assert raw_action_extra["metadata"]["tags"] == ["action"] + + @pytest.mark.parametrize( + ("raw_item", "tool_name", "call_id"), + [ + ( + ResponseCustomToolCall( + type="custom_tool_call", + name="custom_tool", + call_id="cid-custom-decision", + input="original", + ), + "custom_tool", + "cid-custom-decision", + ), + ( + ResponseFunctionShellToolCall.model_validate( + { + "id": "shell-decision", + "action": {"commands": ["echo", "ok"]}, + "call_id": "cid-shell-decision", + "status": "completed", + "type": "shell_call", + } + ), + "shell", + "cid-shell-decision", + ), + ], + ids=["custom", "shell"], + ) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_approval_pipeline_models_can_apply_detached_decisions( + self, + raw_item: Any, + tool_name: str, + call_id: str, + approve: bool, + ) -> None: + """Production approval models must detach and retain canonical routing identity.""" + agent = Agent(name="PipelineDecisionAgent") + approval_item = ToolApprovalItem( + agent=agent, + raw_item=raw_item, + tool_name=tool_name, + ) + state = make_state_with_interruptions(agent, [approval_item]) - def test_get_interruptions_returns_a_snapshot(self): - """Mutating returned interruptions must not change pending approvals.""" - agent = Agent(name="SnapshotAgent") + interruption = state.get_interruptions()[0] + if approve: + state.approve(interruption) + else: + state.reject(interruption) + + assert state._context is not None + assert state._context.is_tool_approved(tool_name, call_id) is approve + + def test_get_interruptions_detaches_a_nested_mutable_alias(self): + """Snapshot copying must not trust a nested object's deepcopy implementation.""" + + class SelfCopyingList(list[str]): + def __deepcopy__(self, _memo: dict[int, Any]) -> SelfCopyingList: + return self + + agent = Agent(name="AliasedAgent") + metadata = SelfCopyingList(["original"]) approval_item = ToolApprovalItem( agent=agent, - raw_item=ResponseFunctionToolCall( - type="function_call", - name="toolA", - call_id="cid-snapshot", - status="completed", - arguments="{}", - ), + raw_item={ + "type": "function_call", + "name": "toolA", + "call_id": "cid-aliased", + "status": "completed", + "arguments": "{}", + "metadata": metadata, + }, ) state = make_state_with_interruptions(agent, [approval_item]) - state.get_interruptions().clear() + interruption = state.get_interruptions()[0] + assert isinstance(interruption.raw_item, dict) + interruption.raw_item["metadata"].append("changed") - assert state.get_interruptions() == [approval_item] + assert metadata == ["original"] async def test_serializes_and_restores_approvals(self): """Test that approval state is preserved through serialization.""" @@ -4029,6 +5325,466 @@ async def test_ambiguous_current_and_nested_approval_identity_fails_closed( scope_id=target_state._agent_tool_state_scope_id, ) + @pytest.mark.parametrize("approval_input", ["snapshot", "exact"]) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_ambiguous_current_approval_identity_fails_closed( + self, + approval_input: str, + approve: bool, + ) -> None: + """A snapshot or exact approval shared by current owners must not be guessed.""" + agent = Agent(name="AmbiguousCurrentAgent") + raw_item = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="shared-current", + arguments="{}", + ) + first = ToolApprovalItem( + agent=agent, + raw_item=raw_item, + tool_name="toolA", + tool_lookup_key=("deferred_top_level", "toolA"), + _allow_bare_name_alias=True, + ) + second = replace( + first, + raw_item=raw_item.model_copy(deep=True), + _allow_bare_name_alias=False, + ) + state = make_state_with_interruptions(agent, [first, second]) + + selected = state.get_interruptions()[1] if approval_input == "snapshot" else second + with pytest.raises(UserError, match="multiple current pending approvals"): + if approve: + state.approve(selected) + else: + state.reject(selected) + + assert state._context is not None + assert state._context.is_tool_approved("toolA", "shared-current") is None + + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_unsafe_current_sibling_cannot_bypass_approval_ambiguity( + self, + approve: bool, + ) -> None: + """Unsafe same-Agent siblings must not be treated as distinct owners.""" + agent = Agent(name="UnsafeCurrentSiblingAgent") + first = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "toolA", + "call_id": "shared-unsafe-current", + "arguments": "{}", + "metadata": object(), + }, + ) + second = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "toolA", + "call_id": "shared-unsafe-current", + "arguments": "{}", + "metadata": object(), + }, + ) + state = make_state_with_interruptions(agent, [first, second]) + + with pytest.raises(UserError, match="multiple current pending approvals"): + if approve: + state.approve(second) + else: + state.reject(second) + + assert state._context is not None + assert state._context.is_tool_approved("toolA", "shared-unsafe-current") is None + + @pytest.mark.parametrize("location", ["current", "nested"]) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_uncopyable_noncanonical_approval_does_not_select_pending_owner( + self, + location: str, + approve: bool, + ) -> None: + """An uncopyable noncanonical input must not select a same-Agent pending owner.""" + from agents.agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result + + agent = Agent(name="UncopyableDecisionAgent") + pending = ToolApprovalItem( + agent=agent, + raw_item=ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="cid-uncopyable-decision", + arguments="{}", + ), + ) + supplied = ToolApprovalItem(agent=agent, raw_item={"metadata": object()}) + + target_state = make_state_with_interruptions(agent, [pending]) + state = target_state + outer_call = make_tool_call( + call_id="outer-uncopyable-decision", + name="nested_agent_tool", + ) + if location == "nested": + state = make_state_with_interruptions(agent, []) + nested_tool = function_tool(lambda: "nested", name_override="nested_agent_tool") + state._last_processed_response = make_processed_response( + functions=[ToolRunFunction(tool_call=outer_call, function_tool=nested_tool)] + ) + record_agent_tool_run_result( + outer_call, + cast( + Any, + SimpleNamespace( + interruptions=[pending], + to_state=lambda: target_state, + ), + ), + scope_id=state._agent_tool_state_scope_id, + ) + + try: + if approve: + state.approve(supplied) + else: + state.reject(supplied) + + assert target_state._context is not None + assert ( + target_state._context.is_tool_approved( + "toolA", + "cid-uncopyable-decision", + ) + is None + ) + finally: + if location == "nested": + drop_agent_tool_run_result( + outer_call, + scope_id=state._agent_tool_state_scope_id, + ) + + @pytest.mark.parametrize("location", ["current", "nested"]) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_safe_noncanonical_approval_does_not_select_uncopyable_pending_owner( + self, + location: str, + approve: bool, + ) -> None: + """A safe input must not be redirected to an unsafe same-Agent pending owner.""" + from agents.agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result + + agent = Agent(name="UnsafePendingOwnerAgent") + pending = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "pending_tool", + "call_id": "pending-unsafe-owner", + "arguments": "{}", + "metadata": object(), + }, + ) + supplied = ToolApprovalItem( + agent=agent, + raw_item=ResponseFunctionToolCall( + type="function_call", + name="supplied_tool", + call_id="supplied-safe-owner", + arguments="{}", + ), + ) + + target_state = make_state_with_interruptions(agent, [pending]) + state = target_state + outer_call = make_tool_call(call_id="outer-unsafe-owner", name="nested_agent_tool") + if location == "nested": + state = make_state_with_interruptions(agent, []) + nested_tool = function_tool(lambda: "nested", name_override="nested_agent_tool") + state._last_processed_response = make_processed_response( + functions=[ToolRunFunction(tool_call=outer_call, function_tool=nested_tool)] + ) + record_agent_tool_run_result( + outer_call, + cast( + Any, + SimpleNamespace( + interruptions=[pending], + to_state=lambda: target_state, + ), + ), + scope_id=state._agent_tool_state_scope_id, + ) + + try: + with pytest.raises(UserError, match="Cannot apply approval"): + if approve: + state.approve(supplied) + else: + state.reject(supplied) + + assert target_state._context is not None + assert ( + target_state._context.is_tool_approved( + "pending_tool", + "pending-unsafe-owner", + ) + is None + ) + assert ( + target_state._context.is_tool_approved( + "supplied_tool", + "supplied-safe-owner", + ) + is None + ) + finally: + if location == "nested": + drop_agent_tool_run_result( + outer_call, + scope_id=state._agent_tool_state_scope_id, + ) + + @pytest.mark.parametrize("approval_location", ["current", "nested"]) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_exact_uncopyable_approval_does_not_read_other_unsafe_owner( + self, + approval_location: str, + approve: bool, + ) -> None: + """An unsafe authoritative item must not expose another owner's raw payload.""" + from agents.agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result + + hook_calls: list[tuple[str, object]] = [] + + class HookedDict(dict[str, Any]): + def get(self, key: str, default: Any = None) -> Any: + hook_calls.append(("get", key)) + return super().get(key, default) + + def __contains__(self, key: object) -> bool: + hook_calls.append(("contains", key)) + return super().__contains__(key) + + def __getitem__(self, key: str) -> Any: + hook_calls.append(("getitem", key)) + return super().__getitem__(key) + + agent = Agent(name="UnsafeAuthoritativeOwnerAgent") + current = ToolApprovalItem( + agent=agent, + raw_item=HookedDict( + type="function_call", + name="current_tool", + call_id="current-unsafe-authoritative", + arguments="{}", + metadata=object(), + ), + ) + nested = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "nested_tool", + "call_id": "nested-unsafe-authoritative", + "arguments": "{}", + "metadata": object(), + }, + ) + outer_state = make_state_with_interruptions(agent, [current]) + nested_state = make_state_with_interruptions(agent, [nested]) + outer_call = make_tool_call( + call_id="outer-unsafe-authoritative", + name="nested_agent_tool", + ) + nested_tool = function_tool(lambda: "nested", name_override="nested_agent_tool") + outer_state._last_processed_response = make_processed_response( + functions=[ToolRunFunction(tool_call=outer_call, function_tool=nested_tool)] + ) + record_agent_tool_run_result( + outer_call, + cast( + Any, + SimpleNamespace( + interruptions=[nested], + to_state=lambda: nested_state, + ), + ), + scope_id=outer_state._agent_tool_state_scope_id, + ) + hook_calls.clear() + approval_item = current if approval_location == "current" else nested + + try: + with pytest.raises(UserError, match="Cannot apply approval"): + if approve: + outer_state.approve(approval_item) + else: + outer_state.reject(approval_item) + + assert hook_calls == [] + assert outer_state._context is not None + assert nested_state._context is not None + assert ( + outer_state._context.is_tool_approved( + "current_tool", + "current-unsafe-authoritative", + ) + is None + ) + assert ( + nested_state._context.is_tool_approved( + "nested_tool", + "nested-unsafe-authoritative", + ) + is None + ) + finally: + drop_agent_tool_run_result( + outer_call, + scope_id=outer_state._agent_tool_state_scope_id, + ) + + @pytest.mark.parametrize("unsafe_metadata", [False, True], ids=["safe", "unsafe"]) + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_ambiguous_exact_nested_approval_identity_fails_closed( + self, + unsafe_metadata: bool, + approve: bool, + ) -> None: + """An exact nested approval must not bypass nested owner multiplicity.""" + from agents.agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result + + agent = Agent(name="AmbiguousNestedAgent") + if unsafe_metadata: + raw_item: Any = { + "type": "function_call", + "name": "toolA", + "call_id": "shared-nested", + "arguments": "{}", + "metadata": object(), + } + second_raw_item: Any = {**raw_item, "metadata": object()} + else: + raw_item = ResponseFunctionToolCall( + type="function_call", + name="toolA", + call_id="shared-nested", + arguments="{}", + ) + second_raw_item = raw_item.model_copy(deep=True) + first = ToolApprovalItem(agent=agent, raw_item=raw_item, tool_name="toolA") + second = replace(first, raw_item=second_raw_item) + nested_state = make_state_with_interruptions(agent, [first, second]) + outer_state = make_state_with_interruptions(agent, [first]) + outer_call = make_tool_call(call_id="outer-ambiguous-nested", name="nested_agent_tool") + nested_tool = function_tool(lambda: "nested", name_override="nested_agent_tool") + outer_state._last_processed_response = make_processed_response( + functions=[ToolRunFunction(tool_call=outer_call, function_tool=nested_tool)] + ) + record_agent_tool_run_result( + outer_call, + cast( + Any, + SimpleNamespace( + interruptions=[first], + to_state=lambda: nested_state, + ), + ), + scope_id=outer_state._agent_tool_state_scope_id, + ) + + try: + with pytest.raises(UserError, match="multiple current pending approvals"): + if approve: + outer_state.approve(first) + else: + outer_state.reject(first) + assert nested_state._context is not None + assert nested_state._context.is_tool_approved("toolA", "shared-nested") is None + finally: + drop_agent_tool_run_result( + outer_call, + scope_id=outer_state._agent_tool_state_scope_id, + ) + + @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) + def test_unsafe_exact_approval_across_nested_states_fails_closed( + self, + approve: bool, + ) -> None: + """Exact unsafe input must preserve ambiguity across all nested owner states.""" + from agents.agent_tool_state import drop_agent_tool_run_result, record_agent_tool_run_result + + agent = Agent(name="UnsafeNestedOwnerAgent") + first = ToolApprovalItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "toolA", + "call_id": "shared-unsafe-nested", + "arguments": "{}", + "metadata": object(), + }, + ) + second = replace(first, raw_item={**first.raw_item, "metadata": object()}) + first_state = make_state_with_interruptions(agent, [first]) + second_state = make_state_with_interruptions(agent, [second]) + outer_state = make_state_with_interruptions(agent, []) + first_outer_call = make_tool_call(call_id="outer-unsafe-first", name="nested_first") + second_outer_call = make_tool_call(call_id="outer-unsafe-second", name="nested_second") + first_tool = function_tool(lambda: "first", name_override="nested_first") + second_tool = function_tool(lambda: "second", name_override="nested_second") + outer_state._last_processed_response = make_processed_response( + functions=[ + ToolRunFunction(tool_call=first_outer_call, function_tool=first_tool), + ToolRunFunction(tool_call=second_outer_call, function_tool=second_tool), + ] + ) + for outer_call, item, nested_state in ( + (first_outer_call, first, first_state), + (second_outer_call, second, second_state), + ): + record_agent_tool_run_result( + outer_call, + cast( + Any, + SimpleNamespace( + interruptions=[item], + to_state=lambda nested_state=nested_state: nested_state, + ), + ), + scope_id=outer_state._agent_tool_state_scope_id, + ) + + try: + with pytest.raises(UserError, match="cannot be safely distinguished"): + if approve: + outer_state.approve(first) + else: + outer_state.reject(first) + + for nested_state in (first_state, second_state): + assert nested_state._context is not None + assert ( + nested_state._context.is_tool_approved( + "toolA", + "shared-unsafe-nested", + ) + is None + ) + finally: + for outer_call in (first_outer_call, second_outer_call): + drop_agent_tool_run_result( + outer_call, + scope_id=outer_state._agent_tool_state_scope_id, + ) + @pytest.mark.parametrize("round_trip", [False, True], ids=["live", "serialized"]) @pytest.mark.parametrize("approve", [True, False], ids=["approve", "reject"]) async def test_completed_current_invocation_does_not_own_nested_approval( diff --git a/tests/test_tool_name_collision_policy.py b/tests/test_tool_name_collision_policy.py index 2d01423034..d67e7b1dba 100644 --- a/tests/test_tool_name_collision_policy.py +++ b/tests/test_tool_name_collision_policy.py @@ -23,8 +23,9 @@ handoff, tool_namespace, ) -from agents.items import ToolCallOutputItem +from agents.items import ToolApprovalItem, ToolCallOutputItem from agents.lifecycle import RunHooks +from agents.run_internal.run_steps import NextStepInterruption from agents.testing import ScriptedModel from agents.tool import Tool, function_tool @@ -37,6 +38,15 @@ def _record(calls: list[str], value: str, result: str | None = None) -> str: return value if result is None else result +def _authoritative_interruption( + state: RunState[Any, Agent[Any]], + call_id: str, +) -> ToolApprovalItem: + """Return the RunState-owned approval used by corruption-path tests.""" + assert isinstance(state._current_step, NextStepInterruption) + return next(item for item in state._current_step.interruptions if item.call_id == call_id) + + @pytest.mark.asyncio async def test_resume_warn_mode_rebinds_queued_mcp_call_to_local_winner() -> None: calls: list[str] = [] @@ -776,7 +786,7 @@ async def test_resume_rejects_conflicting_persisted_identity_before_sibling_effe interruptions = state.get_interruptions() for interruption in interruptions: state.approve(interruption) - conflicting = next(item for item in interruptions if item.call_id == "conflicting_call") + conflicting = _authoritative_interruption(state, "conflicting_call") conflicting.raw_item = { "type": "function_call", "name": "lookup", @@ -812,6 +822,7 @@ async def test_resume_rejects_legacy_approval_name_change_before_side_effects() state = initial_result.to_state() interruption = state.get_interruptions()[0] state.approve(interruption) + interruption = _authoritative_interruption(state, "lookup_call") interruption.tool_lookup_key = None interruption.raw_item = { "type": "function_call", @@ -919,7 +930,7 @@ async def test_approved_malformed_approval_only_stays_pending_without_side_effec assert state._last_processed_response is not None state._last_processed_response.functions = [] state._model_responses[-1] = replace(state._model_responses[-1], output=[]) - interruption.raw_item = malformed_raw_item + _authoritative_interruption(state, "lookup_call").raw_item = malformed_raw_item resumed_result = await Runner.run(agent, state) @@ -966,7 +977,7 @@ async def get_all_tools( state = initial_result.to_state() approval = state.get_interruptions()[0] state.approve(approval) - approval_holder["approval"] = approval + approval_holder["approval"] = _authoritative_interruption(state, "lookup_call") resumed_result = await Runner.run(agent, state) @@ -1005,7 +1016,8 @@ def lookup(amount: int) -> str: state = await RunState.from_json(agent, initial_result.to_state().to_json()) approval = state.get_interruptions()[0] state.approve(approval) - cast(Any, approval.raw_item).arguments = '{"amount":999}' + authoritative = _authoritative_interruption(state, "lookup_call") + cast(Any, authoritative.raw_item).arguments = '{"amount":999}' resumed_result = await Runner.run(agent, state) @@ -1065,7 +1077,7 @@ async def get_all_tools( state._model_responses[-1], output=[program], ) - approval_holder["approval"] = approval + approval_holder["approval"] = _authoritative_interruption(state, "lookup_call") resumed_result = await Runner.run(agent, state) @@ -1116,7 +1128,8 @@ async def get_all_tools( state = initial_result.to_state() approval = state.get_interruptions()[0] state.approve(approval) - cast(Any, approval.raw_item).caller.caller_id = "forged_program" + authoritative = _authoritative_interruption(state, "lookup_call") + cast(Any, authoritative.raw_item).caller.caller_id = "forged_program" assert state._last_processed_response is not None state._last_processed_response.functions = [] state._model_responses[-1] = replace( @@ -1165,7 +1178,7 @@ async def test_resume_rejects_response_backed_approval_lookup_mismatch_before_ef for run in state._last_processed_response.functions if run.tool_call.call_id != "lookup_call" ] - lookup_approval = next(item for item in interruptions if item.call_id == "lookup_call") + lookup_approval = _authoritative_interruption(state, "lookup_call") cast(Any, lookup_approval.raw_item).name = "other" lookup_approval.tool_name = "other" lookup_approval.tool_lookup_key = ("bare", "other") @@ -1380,7 +1393,7 @@ async def test_approved_malformed_queued_approval_stays_pending_without_side_eff interruptions = state.get_interruptions() for interruption in interruptions: state.approve(interruption) - lookup_approval = next(item for item in interruptions if item.call_id == "lookup_call") + lookup_approval = _authoritative_interruption(state, "lookup_call") lookup_approval.raw_item = malformed_raw_item resumed_result = await Runner.run(agent, state) @@ -1417,7 +1430,7 @@ async def test_resume_rejects_cross_kind_approval_identity_before_sibling_effect interruptions = state.get_interruptions() for interruption in interruptions: state.approve(interruption) - lookup_approval = next(item for item in interruptions if item.call_id == "lookup_call") + lookup_approval = _authoritative_interruption(state, "lookup_call") lookup_approval.raw_item = { "type": "custom_tool_call", "name": "evil",