diff --git a/src/agents/run.py b/src/agents/run.py index b3fa3f132d..9d6621e66c 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -77,9 +77,23 @@ snapshot_usage, update_run_state_for_interruption, usage_delta, + validate_output_guardrails_with_server_managed_conversation, validate_session_conversation_settings, ) from .run_internal.approvals import approvals_from_step +from .run_internal.blocked_output import ( + _blocked_output_failure_items, + _BlockedOutputOwnerStarts, + _current_response_boundary, + _final_turn_items_for_persistence, + _has_output_guardrails, + _is_terminal_tool_output_response, + _retained_items_for_blocked_response, + _sanitize_blocked_output_guardrail_results, + _should_defer_interrupted_session_items, + _synchronize_accepted_run_state, + _validate_resumed_session_output_guardrail_safety, +) from .run_internal.error_handlers import ( attach_generic_agent_error, build_run_error_data, @@ -94,7 +108,7 @@ from .run_internal.prompt_cache_key import PromptCacheKeyResolver from .run_internal.run_grouping import resolve_run_grouping_id from .run_internal.run_loop import ( - _retained_items_for_blocked_output, + _safe_redacted_persistence_error, cleanup_models_after_run, finalize_max_turns_handler_output, get_all_tools, @@ -892,6 +906,12 @@ def _mark_response_hooks_started() -> None: current_agent = run_state._current_agent else: current_agent = starting_agent + _validate_resumed_session_output_guardrail_safety( + agent=current_agent, + run_config=run_config, + session=session, + run_state=run_state if is_resumed_state else None, + ) sandbox_runtime.assert_agent_supported(current_agent) should_run_agent_start_hooks = True store_setting = current_agent.model_settings.resolve( @@ -937,6 +957,13 @@ def _mark_response_hooks_started() -> None: try: while True: + validate_output_guardrails_with_server_managed_conversation( + current_agent, + run_config, + conversation_id=conversation_id, + previous_response_id=previous_response_id, + auto_previous_response_id=auto_previous_response_id, + ) if TYPE_CHECKING: # Keep loop-carried types explicit to bound Pyright's flow analysis. original_input = cast( # type: ignore[redundant-cast] @@ -1040,6 +1067,23 @@ def _mark_response_hooks_started() -> None: ) raise UserError("No processed response found in previous state") + resumed_response_boundary = _current_response_boundary( + (), + run_state._last_processed_response, + run_state, + ) + blocked_output_owner_starts = _BlockedOutputOwnerStarts( + nonstreamed_session_items=(resumed_response_boundary.session_start), + run_state_generated_items=( + resumed_response_boundary.generated_start + ), + run_state_session_items=resumed_response_boundary.session_start, + run_state_model_responses=len(run_state._model_responses) - 1, + run_state_tool_output_guardrail_results=len( + run_state._tool_output_guardrail_results + ), + ) + turn_result = await resolve_interrupted_turn( bindings=current_bindings, original_input=original_input, @@ -1049,7 +1093,9 @@ def _mark_response_hooks_started() -> None: hooks=hooks, context_wrapper=context_wrapper, run_config=run_config, - server_manages_conversation=server_conversation_tracker is not None, + server_manages_conversation=( + server_conversation_tracker is not None + ), run_state=run_state, error_handlers=error_handlers, ) @@ -1086,6 +1132,14 @@ def _mark_response_hooks_started() -> None: session_persistence_enabled and turn_session_items and run_state is not None + and not isinstance(turn_result.next_step, NextStepFinalOutput) + and not ( + isinstance(turn_result.next_step, NextStepInterruption) + and _should_defer_interrupted_session_items( + current_agent, + run_config, + ) + ) ): run_state._current_turn_persisted_item_count = ( await save_resumed_turn_items( @@ -1168,13 +1222,119 @@ def _mark_response_hooks_started() -> None: ) if isinstance(turn_result.next_step, NextStepFinalOutput): - await run_output_guardrails( - current_agent.output_guardrails - + (run_config.output_guardrails or []), + if run_state is not None and _has_output_guardrails( + current_agent, run_config + ): + run_state._tool_output_guardrail_results = list( + tool_output_guardrail_results + ) + current_processed_response = ( + turn_result.processed_response + if turn_result.processed_response is not None + else run_state._last_processed_response + ) + output_guardrail_result_start = len(output_guardrail_results) + try: + await run_output_guardrails( + current_agent.output_guardrails + + (run_config.output_guardrails or []), + current_agent, + turn_result.next_step.output, + context_wrapper, + output_guardrail_results, + ) + except OutputGuardrailTripwireTriggered as exc: + if not _is_terminal_tool_output_response( + turn_session_items, + current_processed_response, + run_state, + ): + raise + sanitized_results = _sanitize_blocked_output_guardrail_results( + output_guardrail_results[output_guardrail_result_start:], + exc, + ) + output_guardrail_results[output_guardrail_result_start:] = ( + sanitized_results + ) + session_items = _blocked_output_failure_items( + session_items, + (), + blocked_output_owner_starts, + ) + retained_items = _retained_items_for_blocked_response( + turn_session_items, + turn_result.model_response, + run_state, + current_processed_response, + owner_starts=blocked_output_owner_starts, + ) + list.extend(session_items, retained_items) + try: + await save_final_turn_items_after_guardrails( + session=session, + run_state=run_state, + session_persistence_enabled=( + session_persistence_enabled + ), + input_guardrail_results=( + _attempt_input_guardrail_results() + ), + items=retained_items, + response_id=turn_result.model_response.response_id, + store=store_setting, + wrapper=context_wrapper, + ) + except BaseException as persistence_error: + raise _safe_redacted_persistence_error( + persistence_error + ) from None + raise + except (Exception, asyncio.CancelledError) as guardrail_error: + if not isinstance( + guardrail_error, asyncio.CancelledError + ) or not _is_terminal_tool_output_response( + turn_session_items, + current_processed_response, + run_state, + ): + final_turn_items = _final_turn_items_for_persistence( + turn_session_items, + current_processed_response, + run_state, + current_agent, + run_config, + ) + await save_final_turn_items_after_guardrails( + session=session, + run_state=run_state, + session_persistence_enabled=session_persistence_enabled, + input_guardrail_results=( + _attempt_input_guardrail_results() + ), + items=final_turn_items, + response_id=turn_result.model_response.response_id, + store=store_setting, + wrapper=context_wrapper, + ) + raise + + final_turn_items = _final_turn_items_for_persistence( + turn_session_items, + current_processed_response, + run_state, current_agent, - turn_result.next_step.output, - context_wrapper, - output_guardrail_results, + run_config, + ) + await save_final_turn_items_after_guardrails( + session=session, + run_state=run_state, + session_persistence_enabled=session_persistence_enabled, + input_guardrail_results=_attempt_input_guardrail_results(), + items=final_turn_items, + response_id=turn_result.model_response.response_id, + store=store_setting, + wrapper=context_wrapper, ) current_step = getattr(run_state, "_current_step", None) approvals_from_state = approvals_from_step(current_step) @@ -1202,21 +1362,6 @@ def _mark_response_hooks_started() -> None: ) != list(session_items) if run_state is not None: result._trace_state = run_state._trace_state - if session_persistence_enabled: - input_items_for_save_1: list[TResponseInputItem] = ( - session_input_items_for_persistence - if session_input_items_for_persistence is not None - else [] - ) - await save_result_to_session( - session, - input_items_for_save_1, - session_items_for_turn(turn_result), - run_state, - response_id=turn_result.model_response.response_id, - store=store_setting, - wrapper=context_wrapper, - ) result._original_input = copy_input_items(original_input) run_state._current_step = None return _finalize_result(result) @@ -1393,7 +1538,9 @@ async def _save_max_turns_handler_output( result._original_input = copy_input_items(original_input) return _finalize_result(result) - if run_state is not None and not resuming_turn: + if run_state is not None and ( + not resuming_turn or isinstance(run_state._current_step, NextStepRunAgain) + ): run_state._current_turn_persisted_item_count = 0 logger.debug("Running agent %s (turn %s)", current_agent.name, current_turn) @@ -1406,6 +1553,35 @@ async def _save_max_turns_handler_output( except Exception: last_saved_input_snapshot_for_rewind = None + if run_state is not None and _has_output_guardrails(current_agent, run_config): + _synchronize_accepted_run_state( + run_state, + generated_items=generated_items, + session_items=session_items, + model_responses=model_responses, + tool_input_guardrail_results=tool_input_guardrail_results, + tool_output_guardrail_results=tool_output_guardrail_results, + current_turn=current_turn, + ) + + blocked_output_owner_starts = _BlockedOutputOwnerStarts( + nonstreamed_session_items=len(session_items), + run_state_generated_items=( + len(run_state._generated_items) if run_state is not None else None + ), + run_state_session_items=( + len(run_state._session_items) if run_state is not None else None + ), + run_state_model_responses=( + len(run_state._model_responses) if run_state is not None else None + ), + run_state_tool_output_guardrail_results=( + len(run_state._tool_output_guardrail_results) + if run_state is not None + else None + ), + ) + items_for_model = ( pending_server_items if server_conversation_tracker is not None and pending_server_items @@ -1665,6 +1841,13 @@ async def _save_max_turns_handler_output( try: if isinstance(turn_result.next_step, NextStepFinalOutput): + if run_state is not None and _has_output_guardrails( + current_agent, run_config + ): + run_state._tool_output_guardrail_results = list( + tool_output_guardrail_results + ) + output_guardrail_result_start = len(output_guardrail_results) try: await run_output_guardrails( current_agent.output_guardrails @@ -1674,39 +1857,93 @@ async def _save_max_turns_handler_output( context_wrapper, output_guardrail_results, ) - except OutputGuardrailTripwireTriggered: - await save_final_turn_items_after_guardrails( - session=session, - run_state=run_state, - session_persistence_enabled=session_persistence_enabled, - input_guardrail_results=_attempt_input_guardrail_results(), - items=_retained_items_for_blocked_output(items_to_save_turn), - response_id=turn_result.model_response.response_id, - store=store_setting, - wrapper=context_wrapper, + except OutputGuardrailTripwireTriggered as exc: + if not _is_terminal_tool_output_response( + turn_session_items, + turn_result.processed_response, + run_state, + ): + raise + sanitized_results = _sanitize_blocked_output_guardrail_results( + output_guardrail_results[output_guardrail_result_start:], + exc, ) - raise - except (Exception, asyncio.CancelledError): - # Preserve the released non-stream behavior for guardrail errors - # and cancellation: the completed final turn remains replayable. - await save_final_turn_items_after_guardrails( - session=session, - run_state=run_state, - session_persistence_enabled=session_persistence_enabled, - input_guardrail_results=_attempt_input_guardrail_results(), - items=items_to_save_turn, - response_id=turn_result.model_response.response_id, - store=store_setting, - wrapper=context_wrapper, + output_guardrail_results[output_guardrail_result_start:] = ( + sanitized_results ) + session_items = _blocked_output_failure_items( + session_items, + (), + blocked_output_owner_starts, + ) + retained_items = _retained_items_for_blocked_response( + turn_session_items, + turn_result.model_response, + run_state, + turn_result.processed_response, + owner_starts=blocked_output_owner_starts, + ) + list.extend(session_items, retained_items) + try: + await save_final_turn_items_after_guardrails( + session=session, + run_state=run_state, + session_persistence_enabled=(session_persistence_enabled), + input_guardrail_results=( + _attempt_input_guardrail_results() + ), + items=retained_items, + response_id=turn_result.model_response.response_id, + store=store_setting, + wrapper=context_wrapper, + ) + except BaseException as persistence_error: + raise _safe_redacted_persistence_error( + persistence_error + ) from None + raise + except (Exception, asyncio.CancelledError) as guardrail_error: + if not isinstance( + guardrail_error, asyncio.CancelledError + ) or not _is_terminal_tool_output_response( + turn_session_items, + turn_result.processed_response, + run_state, + ): + final_turn_items = _final_turn_items_for_persistence( + turn_session_items, + turn_result.processed_response, + run_state, + current_agent, + run_config, + ) + await save_final_turn_items_after_guardrails( + session=session, + run_state=run_state, + session_persistence_enabled=session_persistence_enabled, + input_guardrail_results=( + _attempt_input_guardrail_results() + ), + items=final_turn_items, + response_id=turn_result.model_response.response_id, + store=store_setting, + wrapper=context_wrapper, + ) raise + final_turn_items = _final_turn_items_for_persistence( + turn_session_items, + turn_result.processed_response, + run_state, + current_agent, + run_config, + ) await save_final_turn_items_after_guardrails( session=session, run_state=run_state, session_persistence_enabled=session_persistence_enabled, input_guardrail_results=_attempt_input_guardrail_results(), - items=items_to_save_turn, + items=final_turn_items, response_id=turn_result.model_response.response_id, store=store_setting, wrapper=context_wrapper, @@ -1745,7 +1982,12 @@ async def _save_max_turns_handler_output( run_state._current_step = None return _finalize_result(result) elif isinstance(turn_result.next_step, NextStepInterruption): - if session_persistence_enabled: + if session_persistence_enabled and not ( + _should_defer_interrupted_session_items( + current_agent, + run_config, + ) + ): if not input_guardrails_triggered( _attempt_input_guardrail_results() ): @@ -2141,6 +2383,19 @@ def run_streamed( if run_state is not None: run_state._reasoning_item_id_policy = resolved_reasoning_item_id_policy + schema_agent = ( + run_state._current_agent + if run_state is not None and run_state._current_agent is not None + else starting_agent + ) + validate_output_guardrails_with_server_managed_conversation( + schema_agent, + run_config, + conversation_id=conversation_id, + previous_response_id=previous_response_id, + auto_previous_response_id=auto_previous_response_id, + ) + ( trace_workflow_name, trace_id, @@ -2176,11 +2431,6 @@ def run_streamed( run_state=run_state, ) - schema_agent = ( - run_state._current_agent - if run_state is not None and run_state._current_agent is not None - else starting_agent - ) sandbox_runtime.assert_agent_supported(schema_agent) output_schema = get_output_schema(schema_agent) diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py index 7d5b73ad5a..be1d976724 100644 --- a/src/agents/run_internal/agent_runner_helpers.py +++ b/src/agents/run_internal/agent_runner_helpers.py @@ -14,6 +14,7 @@ from ..items import ModelResponse, RunItem, ToolApprovalItem, TResponseInputItem from ..memory import Session from ..models.openai_agent_registration import add_openai_harness_id_to_metadata +from ..models.openai_chatcompletions import OpenAIChatCompletionsModel from ..result import RunResult from ..run_config import ReasoningItemIdPolicy, RunConfig from ..run_context import RunContextWrapper, TContext @@ -42,6 +43,7 @@ ) from .session_persistence import save_result_to_session, save_resumed_turn_items from .tool_use_tracker import AgentToolUseTracker, serialize_tool_use_tracker +from .turn_preparation import get_model __all__ = [ "apply_resumed_conversation_settings", @@ -55,6 +57,7 @@ "finalize_conversation_tracking", "get_unsent_tool_call_ids_for_interrupted_state", "input_guardrails_triggered", + "validate_output_guardrails_with_server_managed_conversation", "validate_session_conversation_settings", "resolve_trace_settings", "resolve_processed_response", @@ -257,6 +260,29 @@ def validate_session_conversation_settings( ) +def validate_output_guardrails_with_server_managed_conversation( + agent: Agent[Any], + run_config: RunConfig, + *, + conversation_id: str | None, + previous_response_id: str | None, + auto_previous_response_id: bool, +) -> None: + """Reject an output-guardrail run whose rejected history cannot be locally replaced.""" + if conversation_id is None and previous_response_id is None and not auto_previous_response_id: + return + if not agent.output_guardrails and not run_config.output_guardrails: + return + if isinstance(get_model(agent, run_config), OpenAIChatCompletionsModel): + # Chat Completions owns its released warn-and-ignore or strict rejection behavior. + return + raise UserError( + "Output guardrails cannot be combined with conversation_id, previous_response_id, " + "or auto_previous_response_id because rejected output cannot be removed from " + "server-managed conversation history." + ) + + def resolve_trace_settings( *, run_state: RunState[TContext] | None, diff --git a/src/agents/run_internal/blocked_output.py b/src/agents/run_internal/blocked_output.py new file mode 100644 index 0000000000..bb1da4ffaa --- /dev/null +++ b/src/agents/run_internal/blocked_output.py @@ -0,0 +1,831 @@ +"""Canonical data-free function-tool payloads rejected by an output guardrail.""" + +from __future__ import annotations + +import dataclasses as _dc +from collections.abc import Sequence +from typing import Any, TypeVar, cast + +from openai.types.responses import ResponseFunctionToolCall +from openai.types.responses.response_function_tool_call import CallerDirect + +from ..agent import Agent +from ..exceptions import ( + AgentsException, + OutputGuardrailTripwireTriggered, + UserError, + _detach_data_redacted_error_traceback, + _mark_error_data_redacted, + _prepare_data_redacted_error, +) +from ..guardrail import GuardrailFunctionOutput, OutputGuardrailResult +from ..items import ModelResponse, RunItem, ToolCallItem, ToolCallOutputItem +from ..memory import Session +from ..result import RunResultStreaming +from ..run_config import RunConfig +from ..run_state import RunState +from ..tool_guardrails import ( + ToolGuardrailFunctionOutput, + ToolInputGuardrailResult, + ToolOutputGuardrailResult, +) +from .run_steps import NextStepInterruption, ProcessedResponse + +OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT = "Output withheld by an output guardrail." + +_RESPONSE_OUTPUT_STATUSES = frozenset({"in_progress", "completed", "incomplete"}) + + +def _exact_dict_field(values: dict[Any, Any], field: str) -> Any: + """Read one exact string key without invoking stored-key equality hooks.""" + for key, value in dict.items(values): + if type(key) is str and str.__eq__(key, field) is True: + return value + return None + + +def _payload_field(raw_item: Any, field: str) -> Any: + """Read an allowlisted field without copying extras or invoking instance hooks.""" + if type(raw_item) is dict: + values = raw_item + elif type(raw_item) is ResponseFunctionToolCall: + values = object.__getattribute__(raw_item, "__dict__") + else: + raise AgentsException("Cannot sanitize an unsupported tool item variant.") + if type(values) is not dict: + raise AgentsException("Cannot sanitize an unsupported tool item representation.") + return _exact_dict_field(values, field) + + +def _required_string(raw_item: Any, field: str) -> str: + value = _payload_field(raw_item, field) + if type(value) is not str or not value: + raise AgentsException(f"Cannot sanitize a function tool item without {field}.") + return value + + +def _copy_optional_string( + sanitized: dict[str, Any], + raw_item: Any, + field: str, +) -> None: + value = _payload_field(raw_item, field) + if value is None: + return + if type(value) is not str or not value: + raise AgentsException(f"Cannot sanitize a function tool item with an invalid {field}.") + sanitized[field] = value + + +def _copy_optional_status(sanitized: dict[str, Any], raw_item: Any) -> None: + status = _payload_field(raw_item, "status") + if status is None: + return + if type(status) is not str or status not in _RESPONSE_OUTPUT_STATUSES: + raise AgentsException("Cannot sanitize a function tool item with an invalid status.") + sanitized["status"] = status + + +def _copy_optional_direct_caller(sanitized: dict[str, Any], raw_item: Any) -> None: + caller = _payload_field(raw_item, "caller") + if caller is None: + return + if type(caller) is CallerDirect: + values = object.__getattribute__(caller, "__dict__") + caller_type = _exact_dict_field(values, "type") if type(values) is dict else None + elif type(caller) is dict: + caller_type = _exact_dict_field(caller, "type") + else: + caller_type = None + if type(caller_type) is str and str.__eq__(caller_type, "direct") is True: + sanitized["caller"] = {"type": "direct"} + return + raise AgentsException("Cannot sanitize a function tool item with a non-direct caller.") + + +def blocked_function_call_payload(raw_item: Any) -> dict[str, Any]: + """Build a provider-valid function call from explicitly allowlisted fields.""" + item_type = _payload_field(raw_item, "type") + if type(item_type) is not str or str.__eq__(item_type, "function_call") is not True: + raise AgentsException("Cannot sanitize an unsupported tool call variant.") + arguments = _payload_field(raw_item, "arguments") + if type(arguments) is not str: + raise AgentsException("Cannot sanitize a function tool item without arguments.") + sanitized: dict[str, Any] = { + "type": "function_call", + "name": _required_string(raw_item, "name"), + "arguments": arguments, + "call_id": _required_string(raw_item, "call_id"), + } + _copy_optional_string(sanitized, raw_item, "id") + _copy_optional_string(sanitized, raw_item, "namespace") + _copy_optional_status(sanitized, raw_item) + _copy_optional_direct_caller(sanitized, raw_item) + try: + validated = ResponseFunctionToolCall(**sanitized) + except Exception: + raise AgentsException("Sanitized function_call is not valid for replay.") from None + return validated.model_dump(exclude_unset=True) + + +def blocked_function_output_payload(raw_item: Any) -> dict[str, Any]: + """Build a replay-valid function output from explicitly allowlisted fields.""" + item_type = _payload_field(raw_item, "type") + if type(item_type) is not str or str.__eq__(item_type, "function_call_output") is not True: + raise AgentsException("Cannot sanitize an unsupported tool output variant.") + sanitized: dict[str, Any] = { + "type": "function_call_output", + "call_id": _required_string(raw_item, "call_id"), + "output": OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + } + _copy_optional_string(sanitized, raw_item, "id") + _copy_optional_status(sanitized, raw_item) + _copy_optional_direct_caller(sanitized, raw_item) + try: + from ..run_state import _deserialize_tool_call_output_raw_item + + restored = _deserialize_tool_call_output_raw_item(sanitized) + except Exception: + raise AgentsException("Sanitized function_call_output is not valid for replay.") from None + if restored is None: + raise AgentsException("Sanitized function_call_output is not valid for replay.") + return sanitized + + +_SIDE_EFFECT_ITEM_TYPES = frozenset({"tool_call_item", "tool_call_output_item"}) + + +def _sanitize_blocked_output_guardrail_results( + results: Sequence[OutputGuardrailResult], + tripwire: OutputGuardrailTripwireTriggered, +) -> list[OutputGuardrailResult]: + """Build data-free guardrail results and detach the tripwire from raw output.""" + sanitized_by_id: dict[int, OutputGuardrailResult] = {} + + def sanitize(result: OutputGuardrailResult) -> OutputGuardrailResult: + existing = sanitized_by_id.get(id(result)) + if existing is not None: + return existing + sanitized = OutputGuardrailResult( + guardrail=result.guardrail, + agent_output=OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + agent=result.agent, + output=GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=result.output.tripwire_triggered, + ), + ) + sanitized_by_id[id(result)] = sanitized + return sanitized + + sanitized_results = [sanitize(result) for result in results] + object.__setattr__(tripwire, "guardrail_result", sanitize(tripwire.guardrail_result)) + _mark_error_data_redacted(tripwire) + _detach_data_redacted_error_traceback(tripwire) + return sanitized_results + + +@_dc.dataclass(frozen=True) +class _CurrentResponseBoundary: + """A current-response suffix proven only by lifecycle position or object identity.""" + + items: tuple[RunItem, ...] + processed_items: tuple[RunItem, ...] + generated_start: int | None + session_start: int | None + proven: bool + + +@_dc.dataclass(frozen=True) +class _BlockedOutputSnapshot: + """Prepared data-free replacements for one complete current response.""" + + items: tuple[RunItem, ...] + processed_items: tuple[RunItem, ...] + model_response: ModelResponse | None + + +@_dc.dataclass(frozen=True) +class _BlockedOutputOwnerPlan: + """Prebuilt trusted-owner assignments for application or emergency cleanup.""" + + assignments: tuple[tuple[Any, str, Any], ...] + + +@_dc.dataclass(frozen=True) +class _BlockedOutputOwnerStarts: + """Owner-specific current-response starts captured at trusted lifecycle boundaries.""" + + nonstreamed_session_items: int | None = None + run_state_generated_items: int | None = None + run_state_session_items: int | None = None + run_state_model_responses: int | None = None + run_state_tool_output_guardrail_results: int | None = None + streamed_new_items: int | None = None + streamed_model_input_items: int | None = None + streamed_raw_responses: int | None = None + streamed_tool_output_guardrail_results: int | None = None + + +@_dc.dataclass(frozen=True) +class _BlockedOutputOwnerPrefixes: + """Accepted owner prefixes allocated before any blocked-output replacement begins.""" + + run_state_generated_items: list[RunItem] + run_state_session_items: list[RunItem] + run_state_model_responses: list[ModelResponse] + run_state_tool_output_guardrail_results: list[ToolOutputGuardrailResult] + streamed_new_items: list[RunItem] + streamed_model_input_items: list[RunItem] + streamed_raw_responses: list[ModelResponse] + streamed_tool_output_guardrail_results: list[ToolOutputGuardrailResult] + + +_OwnerItemT = TypeVar("_OwnerItemT") + + +def _has_output_guardrails(agent: Agent[Any], run_config: RunConfig) -> bool: + return bool(agent.output_guardrails or run_config.output_guardrails) + + +def _synchronize_accepted_run_state( + run_state: RunState[Any], + *, + generated_items: Sequence[RunItem], + session_items: Sequence[RunItem], + model_responses: Sequence[ModelResponse], + tool_input_guardrail_results: Sequence[ToolInputGuardrailResult], + tool_output_guardrail_results: Sequence[ToolOutputGuardrailResult], + current_turn: int, +) -> None: + """Capture accepted run history before a guardrail-owned model response begins.""" + run_state._generated_items = list(generated_items) + run_state._session_items = list(session_items) + run_state._model_responses = list(model_responses) + run_state._tool_input_guardrail_results = list(tool_input_guardrail_results) + run_state._tool_output_guardrail_results = list(tool_output_guardrail_results) + run_state._current_turn = current_turn + + +def _should_defer_interrupted_session_items( + agent: Agent[Any], + run_config: RunConfig, +) -> bool: + """Defer only approval state that could still become guarded terminal tool output.""" + return _has_output_guardrails(agent, run_config) and agent.tool_use_behavior != "run_llm_again" + + +def _validate_resumed_session_output_guardrail_safety( + *, + agent: Agent[Any], + run_config: RunConfig, + session: Session | None, + run_state: RunState[Any] | None, +) -> None: + """Reject approval resumes whose current-response boundary is not structurally provable.""" + if run_state is None or not _has_output_guardrails(agent, run_config): + return + if not isinstance(run_state._current_step, NextStepInterruption): + return + boundary = _current_response_boundary( + (), + run_state._last_processed_response, + run_state, + ) + if not boundary.proven: + raise UserError( + "Cannot resume a serialized approval checkpoint with output guardrails because the " + "current response boundary cannot be proven. Start a new run from safe input." + ) + if run_state._current_turn_persisted_item_count > 0 and ( + _should_defer_interrupted_session_items(agent, run_config) + ): + if session is not None: + raise UserError( + "Cannot resume an approval checkpoint with output guardrails after current-turn " + "items were persisted. Start a new run from safe input." + ) + # A detached Session cannot contribute its old persisted prefix to this run. + run_state._current_turn_persisted_item_count = 0 + + +def _identity_sequence_start( + container: Sequence[RunItem], + sequence: Sequence[RunItem], +) -> int | None: + if not sequence or len(sequence) > len(container): + return None + for start in range(len(container) - len(sequence) + 1): + if all(container[start + offset] is item for offset, item in enumerate(sequence)): + return start + return None + + +def _current_response_boundary( + new_items: Sequence[RunItem], + processed_response: ProcessedResponse | None, + run_state: RunState[Any] | None, +) -> _CurrentResponseBoundary: + """Collect one response using only SDK lifecycle position and exact object identity.""" + processed_items = tuple(processed_response.new_items) if processed_response is not None else () + supplied_items = tuple(new_items) + supplied_start = _identity_sequence_start(supplied_items, processed_items) + response_items = ( + supplied_items[supplied_start:] if supplied_start is not None else supplied_items + ) + generated_start = None + session_start = None + proven = run_state is None or not processed_items or supplied_start is not None + suffixes: list[RunItem] = [] + if run_state is not None: + anchor_items = processed_items or response_items + if anchor_items: + generated_start = _identity_sequence_start(run_state._generated_items, anchor_items) + session_start = _identity_sequence_start(run_state._session_items, anchor_items) + if generated_start is not None: + suffixes.extend(run_state._generated_items[generated_start:]) + proven = True + if session_start is not None: + suffixes.extend(run_state._session_items[session_start:]) + proven = True + if generated_start is None and session_start is None and run_state._current_turn == 1: + current_response_prefix = tuple(run_state._generated_items[: len(processed_items)]) + if len(current_response_prefix) == len(processed_items) and all( + type(actual) is type(expected) + for actual, expected in zip(current_response_prefix, processed_items, strict=False) + ): + # Serialization rebuilds item identities, but turn one has no accepted prefix. + processed_items = current_response_prefix + generated_start = 0 + session_start = 0 + suffixes.extend(run_state._generated_items) + suffixes.extend(run_state._session_items) + proven = True + + current_items: list[RunItem] = [] + seen: set[int] = set() + for item in (*processed_items, *suffixes, *response_items): + if id(item) in seen: + continue + seen.add(id(item)) + current_items.append(item) + return _CurrentResponseBoundary( + items=tuple(current_items), + processed_items=processed_items, + generated_start=generated_start, + session_start=session_start, + proven=proven, + ) + + +def _current_response_items_for_persistence( + items: Sequence[RunItem], + processed_response: ProcessedResponse | None, + run_state: RunState[Any] | None = None, +) -> list[RunItem]: + """Return the complete current response or fail before using an ambiguous boundary.""" + boundary = _current_response_boundary(items, processed_response, run_state) + if not boundary.proven: + raise UserError( + "Cannot persist an ambiguous resumed response with output guardrails. " + "Start a new run from safe input." + ) + return list(boundary.items) + + +def _final_turn_items_for_persistence( + items: Sequence[RunItem], + processed_response: ProcessedResponse | None, + run_state: RunState[Any] | None, + agent: Agent[Any], + run_config: RunConfig, +) -> list[RunItem]: + """Use released resumed suffix persistence unless output guardrails defer the response.""" + if not _has_output_guardrails(agent, run_config): + return list(items) + return _current_response_items_for_persistence(items, processed_response, run_state) + + +def _is_terminal_tool_output_response( + items: Sequence[RunItem], + processed_response: ProcessedResponse | None, + run_state: RunState[Any] | None = None, +) -> bool: + """Return whether the structurally owned current response produced a tool final output.""" + boundary = _current_response_boundary(items, processed_response, run_state) + return boundary.proven and any(isinstance(item, ToolCallOutputItem) for item in boundary.items) + + +def _prepare_blocked_output_snapshot( + boundary: _CurrentResponseBoundary, + model_response: ModelResponse | None, +) -> _BlockedOutputSnapshot: + """Build an allowlist-only function call/output snapshot before changing live state.""" + current_items = list(boundary.items) + if any(item.type == "reasoning_item" for item in current_items): + raise AgentsException("Cannot sanitize a response containing reasoning items.") + retained_indexes = { + index for index, item in enumerate(current_items) if item.type in _SIDE_EFFECT_ITEM_TYPES + } + replacements: dict[int, RunItem] = {} + calls_by_id: dict[str, int] = {} + outputs_by_id: dict[str, int] = {} + for index in sorted(retained_indexes): + item = current_items[index] + if isinstance(item, ToolCallItem): + payload = blocked_function_call_payload(item.raw_item) + call_id = cast(str, payload["call_id"]) + if call_id in calls_by_id: + raise AgentsException("Cannot sanitize duplicate function calls.") + calls_by_id[call_id] = index + replacements[index] = ToolCallItem( + agent=item.agent, + raw_item=cast(Any, payload), + description=item.description, + title=item.title, + tool_origin=item.tool_origin, + _resolved_tool_name=item._resolved_tool_name, + ) + elif isinstance(item, ToolCallOutputItem): + payload = blocked_function_output_payload(item.raw_item) + call_id = cast(str, payload["call_id"]) + if call_id in outputs_by_id: + raise AgentsException("Cannot sanitize duplicate function outputs.") + outputs_by_id[call_id] = index + replacements[index] = ToolCallOutputItem( + agent=item.agent, + raw_item=cast(Any, payload), + output=OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + tool_origin=item.tool_origin, + custom_data=None, + ) + else: + raise AgentsException("Cannot sanitize an unsupported side-effect item.") + + if not outputs_by_id or set(outputs_by_id) - set(calls_by_id): + raise AgentsException("Cannot sanitize an incomplete function call/output batch.") + retained_indexes = { + index + for call_id in outputs_by_id + for index in (calls_by_id[call_id], outputs_by_id[call_id]) + } + retained_items = tuple(replacements[index] for index in sorted(retained_indexes)) + processed_indexes = {id(item): index for index, item in enumerate(current_items)} + retained_processed_items = tuple( + replacements.get(processed_indexes[id(item)], item) + for item in boundary.processed_items + if processed_indexes.get(id(item)) in retained_indexes + ) + sanitized_response = None + if model_response is not None: + sanitized_response = ModelResponse( + output=cast(Any, [item.raw_item for item in retained_processed_items]), + usage=model_response.usage, + response_id=model_response.response_id, + request_id=model_response.request_id, + raw_usage=model_response.raw_usage, + ) + return _BlockedOutputSnapshot( + items=retained_items, + processed_items=retained_processed_items, + model_response=sanitized_response, + ) + + +def _blocked_output_owner_prefix(items: list[_OwnerItemT], start: int | None) -> list[_OwnerItemT]: + """Copy a structurally captured prefix without consulting item values or identities.""" + if start is None or start < 0 or start > len(items): + return [] + return list.__getitem__(items, slice(0, start)) + + +def _blocked_output_failure_items( + items: list[RunItem], + retained_items: Sequence[RunItem], + owner_starts: _BlockedOutputOwnerStarts, +) -> list[RunItem]: + """Build the non-streamed accepted prefix plus the data-free current response.""" + return [ + *_blocked_output_owner_prefix(items, owner_starts.nonstreamed_session_items), + *retained_items, + ] + + +def _prepare_blocked_output_owner_prefixes( + run_state: RunState[Any] | None, + streamed_result: RunResultStreaming | None, + owner_starts: _BlockedOutputOwnerStarts, +) -> _BlockedOutputOwnerPrefixes: + """Allocate every accepted owner prefix before snapshot application begins.""" + return _BlockedOutputOwnerPrefixes( + run_state_generated_items=( + _blocked_output_owner_prefix( + run_state._generated_items, + owner_starts.run_state_generated_items, + ) + if run_state is not None + else [] + ), + run_state_session_items=( + _blocked_output_owner_prefix( + run_state._session_items, + owner_starts.run_state_session_items, + ) + if run_state is not None + else [] + ), + run_state_model_responses=( + _blocked_output_owner_prefix( + run_state._model_responses, + owner_starts.run_state_model_responses, + ) + if run_state is not None + else [] + ), + run_state_tool_output_guardrail_results=( + _blocked_output_owner_prefix( + run_state._tool_output_guardrail_results, + owner_starts.run_state_tool_output_guardrail_results, + ) + if run_state is not None + else [] + ), + streamed_new_items=( + _blocked_output_owner_prefix( + streamed_result.new_items, + owner_starts.streamed_new_items, + ) + if streamed_result is not None + else [] + ), + streamed_model_input_items=( + _blocked_output_owner_prefix( + streamed_result._model_input_items, + owner_starts.streamed_model_input_items, + ) + if streamed_result is not None + else [] + ), + streamed_raw_responses=( + _blocked_output_owner_prefix( + streamed_result.raw_responses, + owner_starts.streamed_raw_responses, + ) + if streamed_result is not None + else [] + ), + streamed_tool_output_guardrail_results=( + _blocked_output_owner_prefix( + streamed_result.tool_output_guardrail_results, + owner_starts.streamed_tool_output_guardrail_results, + ) + if streamed_result is not None + else [] + ), + ) + + +def _prepare_blocked_output_cleanup_plan( + run_state: RunState[Any] | None, + streamed_result: RunResultStreaming | None, + prefixes: _BlockedOutputOwnerPrefixes, +) -> _BlockedOutputOwnerPlan: + """Prepare accepted-prefix cleanup containers before snapshot application begins.""" + assignments: list[tuple[Any, str, Any]] = [] + if run_state is not None: + assignments.extend( + [ + (run_state, "_generated_items", prefixes.run_state_generated_items), + (run_state, "_session_items", prefixes.run_state_session_items), + (run_state, "_model_responses", prefixes.run_state_model_responses), + (run_state, "_last_processed_response", None), + (run_state, "_current_step", None), + (run_state, "_generated_items_last_processed_marker", None), + ( + run_state, + "_tool_output_guardrail_results", + prefixes.run_state_tool_output_guardrail_results, + ), + ] + ) + if streamed_result is not None: + assignments.extend( + [ + (streamed_result, "new_items", prefixes.streamed_new_items), + (streamed_result, "raw_responses", prefixes.streamed_raw_responses), + ( + streamed_result, + "_model_input_items", + prefixes.streamed_model_input_items, + ), + (streamed_result, "_last_processed_response", None), + ( + streamed_result, + "tool_output_guardrail_results", + prefixes.streamed_tool_output_guardrail_results, + ), + ] + ) + return _BlockedOutputOwnerPlan(assignments=tuple(assignments)) + + +def _sever_blocked_output_replay_graph(cleanup_plan: _BlockedOutputOwnerPlan) -> None: + """Best-effort leaf cleanup using only containers allocated before application.""" + for owner, field, value in cleanup_plan.assignments: + try: + object.__setattr__(owner, field, value) + except BaseException: + continue + + +def _data_free_tool_output_guardrail_results( + results: Sequence[ToolOutputGuardrailResult], +) -> tuple[ToolOutputGuardrailResult, ...]: + """Rebuild current-turn tool guardrail results without retaining caller output data.""" + replacements: list[ToolOutputGuardrailResult] = [] + try: + for result in results: + if not isinstance(result, ToolOutputGuardrailResult): + return () + original_output = object.__getattribute__(result, "output") + behavior = object.__getattribute__(original_output, "behavior") + if type(behavior) is not dict: + return () + behavior_type = _exact_dict_field(behavior, "type") + if type(behavior_type) is not str: + return () + if str.__eq__(behavior_type, "allow") is True: + sanitized_output = ToolGuardrailFunctionOutput.allow( + output_info=OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + ) + elif str.__eq__(behavior_type, "reject_content") is True: + sanitized_output = ToolGuardrailFunctionOutput.reject_content( + message=OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + output_info=OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + ) + elif str.__eq__(behavior_type, "raise_exception") is True: + sanitized_output = ToolGuardrailFunctionOutput.raise_exception( + output_info=OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + ) + else: + return () + replacements.append( + ToolOutputGuardrailResult( + guardrail=object.__getattribute__(result, "guardrail"), + output=sanitized_output, + ) + ) + except Exception: + return () + return tuple(replacements) + + +def _prepare_blocked_output_owner_plan( + boundary: _CurrentResponseBoundary, + snapshot: _BlockedOutputSnapshot | None, + model_response: ModelResponse | None, + run_state: RunState[Any] | None, + streamed_result: RunResultStreaming | None, + prefixes: _BlockedOutputOwnerPrefixes, + cleanup_plan: _BlockedOutputOwnerPlan, +) -> _BlockedOutputOwnerPlan: + """Build every owner replacement before applying any of them.""" + safe_items = list(snapshot.items) if snapshot is not None else [] + safe_response = snapshot.model_response if snapshot is not None else None + assignments: list[tuple[Any, str, Any]] = [] + if streamed_result is not None: + public_results = streamed_result.tool_output_guardrail_results + current_results = list.__getitem__( + public_results, + slice(len(prefixes.streamed_tool_output_guardrail_results), None), + ) + elif run_state is not None: + current_results = list.__getitem__( + run_state._tool_output_guardrail_results, + slice(len(prefixes.run_state_tool_output_guardrail_results), None), + ) + else: + current_results = [] + safe_tool_output_guardrail_results = _data_free_tool_output_guardrail_results(current_results) + if streamed_result is not None: + public_safe_results = [ + *prefixes.streamed_tool_output_guardrail_results, + *safe_tool_output_guardrail_results, + ] + else: + public_safe_results = [] + + if run_state is not None: + if boundary.proven: + responses = [ + *prefixes.run_state_model_responses, + *( + [safe_response] + if model_response is not None and safe_response is not None + else [] + ), + ] + run_state_safe_results = [ + *prefixes.run_state_tool_output_guardrail_results, + *safe_tool_output_guardrail_results, + ] + assignments.extend( + [ + ( + run_state, + "_generated_items", + [*prefixes.run_state_generated_items, *safe_items], + ), + ( + run_state, + "_session_items", + [*prefixes.run_state_session_items, *safe_items], + ), + (run_state, "_model_responses", responses), + (run_state, "_last_processed_response", None), + (run_state, "_current_step", None), + (run_state, "_generated_items_last_processed_marker", None), + (run_state, "_tool_output_guardrail_results", run_state_safe_results), + ] + ) + else: + return cleanup_plan + + if streamed_result is not None: + responses = [ + *prefixes.streamed_raw_responses, + *([safe_response] if model_response is not None and safe_response is not None else []), + ] + assignments.extend( + [ + ( + streamed_result, + "new_items", + [*prefixes.streamed_new_items, *safe_items], + ), + ( + streamed_result, + "_model_input_items", + [*prefixes.streamed_model_input_items, *safe_items], + ), + (streamed_result, "raw_responses", responses), + (streamed_result, "_last_processed_response", None), + ( + streamed_result, + "tool_output_guardrail_results", + public_safe_results, + ), + ] + ) + return _BlockedOutputOwnerPlan(assignments=tuple(assignments)) + + +def _apply_blocked_output_owner_plan(plan: _BlockedOutputOwnerPlan) -> None: + """Apply only values that were fully constructed before the first owner swap.""" + for owner, field, value in plan.assignments: + object.__setattr__(owner, field, value) + + +def _retained_items_for_blocked_response( + items: list[RunItem], + model_response: ModelResponse | None, + run_state: RunState[Any] | None = None, + processed_response: ProcessedResponse | None = None, + streamed_result: RunResultStreaming | None = None, + owner_starts: _BlockedOutputOwnerStarts | None = None, +) -> list[RunItem]: + """Return a complete data-free response or discard the entire unsupported suffix.""" + boundary = _current_response_boundary(items, processed_response, run_state) + prefixes = _prepare_blocked_output_owner_prefixes( + run_state, + streamed_result, + owner_starts if owner_starts is not None else _BlockedOutputOwnerStarts(), + ) + cleanup_plan = _prepare_blocked_output_cleanup_plan(run_state, streamed_result, prefixes) + snapshot: _BlockedOutputSnapshot | None = None + try: + if boundary.proven: + snapshot = _prepare_blocked_output_snapshot(boundary, model_response) + except Exception: + snapshot = None + except BaseException: + _sever_blocked_output_replay_graph(cleanup_plan) + raise + try: + owner_plan = _prepare_blocked_output_owner_plan( + boundary, + snapshot, + model_response, + run_state, + streamed_result, + prefixes, + cleanup_plan, + ) + _apply_blocked_output_owner_plan(owner_plan) + except Exception as error: + _sever_blocked_output_replay_graph(cleanup_plan) + raise _prepare_data_redacted_error(error) from None + except BaseException: + _sever_blocked_output_replay_graph(cleanup_plan) + raise + return list(snapshot.items) if snapshot is not None else [] diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 90a1320a84..6125e56d8f 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -84,6 +84,7 @@ Tool, dispose_resolved_computers, ) +from ..tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult from ..tracing import Span, SpanError, agent_span, get_current_trace, task_span, turn_span from ..tracing.config import include_task_and_turn_spans from ..tracing.model_tracing import get_model_tracing_impl @@ -103,8 +104,22 @@ get_unsent_tool_call_ids_for_interrupted_state, snapshot_usage, usage_delta, + validate_output_guardrails_with_server_managed_conversation, ) from .approvals import approvals_from_step +from .blocked_output import ( + OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + _BlockedOutputOwnerStarts, + _current_response_boundary, + _final_turn_items_for_persistence, + _has_output_guardrails, + _is_terminal_tool_output_response, + _retained_items_for_blocked_response, + _sanitize_blocked_output_guardrail_results, + _should_defer_interrupted_session_items, + _synchronize_accepted_run_state, + _validate_resumed_session_output_guardrail_safety, +) from .error_handlers import ( attach_generic_agent_error, build_run_error_data, @@ -273,6 +288,7 @@ "input_guardrail_tripwire_triggered_for_stream", ] +_OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT = OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT _STREAM_EVENT_ITEM_OCCURRENCE_KEY = "_agents_stream_event_item_occurrence_key" @@ -450,70 +466,23 @@ async def _run_output_guardrails_for_stream( # Publish at a single boundary so no failure path can omit results that already # finished. A guardrail raising a non-tripwire error reports the same completed # results a tripwire does. + if not isinstance(exc, OutputGuardrailTripwireTriggered): + log_model_action_error(logger, "Unexpected error in output guardrails", exc) streamed_result.output_guardrail_results = ( streamed_result.output_guardrail_results + completed_results ) - if not isinstance(exc, OutputGuardrailTripwireTriggered): - log_model_action_error(logger, "Unexpected error in output guardrails", exc) raise -_SIDE_EFFECT_ITEM_TYPES = frozenset({"tool_call_item", "tool_call_output_item"}) - - -def _reasoning_indexes_tied_to_retained_items( +def _retained_items_for_blocked_output( items: list[RunItem], - retained_indexes: set[int], -) -> set[int]: - """Indexes of the reasoning items whose tied item is being retained. - - Applies the same association rule as - ``agents.run_internal.items._drop_reasoning_items_preceding_dropped_calls``: a reasoning item - is tied to the next *non-reasoning* model-emitted item. Keeping a group whose following item is - dropped would leave a dangling reasoning item, which the Responses API rejects on the next - request (``reasoning was provided without its required following item``); dropping a group - whose following item is retained would strip the context that call needs to be replayed. - - A trailing reasoning group - one with no following non-reasoning item at all - is not tied to - anything retained, so it is dropped. Note this is stricter than the reference, which keeps such - a group because the item it belongs to may still arrive later in a longer history; here the - turn is complete, so there is nothing left to tie it to. - """ - tied: set[int] = set() - for index in range(len(items) - 1, -1, -1): - if items[index].type != "reasoning_item": - continue - for next_index in range(index + 1, len(items)): - if items[next_index].type == "reasoning_item": - continue - if next_index in retained_indexes: - tied.add(index) - break - return tied - - -def _retained_items_for_blocked_output(items: list[RunItem]) -> list[RunItem]: - """Pick out the items of a final turn to keep when its output is not deliverable. - - A tool that already ran has to stay in the session, together with the context needed to replay - its call. Everything else - the assistant message the guardrail rejected above all - is dropped, - including the reasoning that belongs to the rejected message rather than to a retained call. - - ``_SIDE_EFFECT_ITEM_TYPES`` is enumerated rather than derived, so an item type added later is - *discarded* here by default and has to be classified deliberately. A record of a side effect - that goes unclassified is a bug, so the safer default is the one that surfaces as a missing item - rather than as a rejected message quietly reaching the session. - """ - retained_indexes = { - index for index, item in enumerate(items) if item.type in _SIDE_EFFECT_ITEM_TYPES - } - if not retained_indexes: - return [] - # Reasoning items are not side effects themselves, but a reasoning model requires the reasoning - # item tied to a function call to accompany it in the next request. - retained_indexes |= _reasoning_indexes_tied_to_retained_items(items, retained_indexes) - # Indexed rather than filtered by type so the retained items keep the model's own order. - return [item for index, item in enumerate(items) if index in retained_indexes] + model_response: ModelResponse | None = None, +) -> list[RunItem]: + """Return trusted retained items without consulting earlier provider identities.""" + return _retained_items_for_blocked_response( + items, + model_response, + ) async def _finalize_streamed_final_output( @@ -525,17 +494,15 @@ async def _finalize_streamed_final_output( context_wrapper: RunContextWrapper[TContext], save_items: Callable[[list[RunItem], str | None, bool | None], Awaitable[None]], items: list[RunItem], + model_response: ModelResponse | None, + processed_response: ProcessedResponse | None, + owner_starts: _BlockedOutputOwnerStarts, response_id: str | None, store_setting: bool | None, - persist_before_output_guardrails: bool, on_persisted_after_guardrails: Callable[[bool], None] | None = None, ) -> None: + output_guardrail_result_start = len(streamed_result.output_guardrail_results) redacted_persistence_error: BaseException | None = None - if persist_before_output_guardrails: - # A resumed approval has already committed the tool side effect, so keep its call/output - # pair even when an agent output guardrail blocks delivery of the final result. - await save_items(items, response_id, store_setting) - try: output_guardrail_results = await _run_output_guardrails_for_stream( agent=agent, @@ -544,62 +511,85 @@ async def _finalize_streamed_final_output( context_wrapper=context_wrapper, streamed_result=streamed_result, ) - except OutputGuardrailTripwireTriggered: + except OutputGuardrailTripwireTriggered as exc: + if not _is_terminal_tool_output_response( + items, + processed_response, + streamed_result._state, + ): + raise # The blocked output itself is not persisted, but a tool that already ran is: the next run # has to see that side effect rather than re-issue it. This turn reaches here with tool # items when `tool_use_behavior="stop_on_first_tool"` (or `stop_at_tool_names`, or a custom # callable) turned a tool result straight into the final output. - if not persist_before_output_guardrails: - retained_items = _retained_items_for_blocked_output(items) - if retained_items: + sanitized_results = _sanitize_blocked_output_guardrail_results( + streamed_result.output_guardrail_results[output_guardrail_result_start:], + exc, + ) + streamed_result.output_guardrail_results = [ + *streamed_result.output_guardrail_results[:output_guardrail_result_start], + *sanitized_results, + ] + retained_items = _retained_items_for_blocked_response( + items, + model_response, + streamed_result._state, + processed_response, + streamed_result, + owner_starts, + ) + if retained_items: + try: await save_items(retained_items, response_id, store_setting) + except BaseException as persistence_error: + safe_error = _safe_redacted_persistence_error(persistence_error) + if ( + isinstance(persistence_error, asyncio.CancelledError) + and streamed_result._cancel_mode != "immediate" + ): + streamed_result._stored_exception = safe_error + streamed_result.is_complete = True + streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) + return + raise safe_error from None raise except Exception as guardrail_error: - # Only a tripwire means the output was judged undeliverable. A guardrail error leaves the - # verdict unknown, so the completed final turn is persisted whole and remains replayable. - # `asyncio.CancelledError` is deliberately not caught here: `cancel()` in its default - # immediate mode has to stay prompt, and awaiting a session write would block - # `stream_events()` on an arbitrary backend. `after_turn` is the mode that finishes the - # turn and saves. guardrail_error_is_redacted = _is_error_data_redacted(guardrail_error) if guardrail_error_is_redacted: _detach_data_redacted_error_traceback(guardrail_error) - if not persist_before_output_guardrails: - try: - await save_items(items, response_id, store_setting) - except BaseException as persistence_error: - if guardrail_error_is_redacted: - safe_persistence_error = _safe_redacted_persistence_error(persistence_error) - if ( - isinstance(safe_persistence_error, asyncio.CancelledError) - and streamed_result._cancel_mode != "immediate" - ): - # A cancelled session write is distinct from the caller requesting - # immediate cancellation. Retain a safe cancellation for `stream_events()` - # without completing the run-loop task with the payload-bearing backend - # exception. - streamed_result._stored_exception = safe_persistence_error - streamed_result.is_complete = True - streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) - return - if isinstance(safe_persistence_error, asyncio.CancelledError): - # Public immediate cancellation already owns stream completion and must - # not surface a recovery failure. - return - redacted_persistence_error = safe_persistence_error + try: + final_turn_items = _final_turn_items_for_persistence( + items, + processed_response, + streamed_result._state, + agent, + run_config, + ) + await save_items(final_turn_items, response_id, store_setting) + except BaseException as persistence_error: + if guardrail_error_is_redacted: + safe_persistence_error = _safe_redacted_persistence_error(persistence_error) if ( - isinstance(persistence_error, asyncio.CancelledError) + isinstance(safe_persistence_error, asyncio.CancelledError) and streamed_result._cancel_mode != "immediate" ): - # A cancelled session write is distinct from the caller requesting immediate - # cancellation. The run-loop task itself becomes cancelled, so retain the - # backend cancellation for `stream_events()` to surface. - streamed_result._stored_exception = persistence_error - if redacted_persistence_error is None: - raise - else: - if on_persisted_after_guardrails is not None: - on_persisted_after_guardrails(False) + streamed_result._stored_exception = safe_persistence_error + streamed_result.is_complete = True + streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) + return + if isinstance(safe_persistence_error, asyncio.CancelledError): + return + redacted_persistence_error = safe_persistence_error + if ( + isinstance(persistence_error, asyncio.CancelledError) + and streamed_result._cancel_mode != "immediate" + ): + streamed_result._stored_exception = persistence_error + if redacted_persistence_error is None: + raise + else: + if on_persisted_after_guardrails is not None: + on_persisted_after_guardrails(False) if redacted_persistence_error is None: raise @@ -607,23 +597,29 @@ async def _finalize_streamed_final_output( raise redacted_persistence_error from None streamed_result.output_guardrail_results.extend(output_guardrail_results) + final_turn_items = _final_turn_items_for_persistence( + items, + processed_response, + streamed_result._state, + agent, + run_config, + ) - if not persist_before_output_guardrails: - # Saved as one ordered batch so the session mirrors the model response. Doing it in two - # halves would both reorder the turn and, because the first save advances the turn's - # persisted-item count, make the second one a no-op. - if on_persisted_after_guardrails is None: - await save_items(items, response_id, store_setting) - else: - try: - await save_items(items, response_id, store_setting) - except asyncio.CancelledError as persistence_error: - if streamed_result._cancel_mode == "immediate": - raise - streamed_result._stored_exception = persistence_error - streamed_result.is_complete = True - streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) - return + # Saved as one ordered batch so the session mirrors the model response. Doing it in two + # halves would both reorder the turn and, because the first save advances the turn's + # persisted-item count, make the second one a no-op. + if on_persisted_after_guardrails is None: + await save_items(final_turn_items, response_id, store_setting) + else: + try: + await save_items(final_turn_items, response_id, store_setting) + except asyncio.CancelledError as persistence_error: + if streamed_result._cancel_mode == "immediate": + raise + streamed_result._stored_exception = persistence_error + streamed_result.is_complete = True + streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) + return streamed_result.final_output = output if on_persisted_after_guardrails is not None: @@ -808,6 +804,9 @@ async def _persist_stream_input_if_needed( def _accumulate_tool_guardrail_results( streamed_result: RunResultStreaming, turn_result: SingleStepResult, + *, + accepted_input_results: list[ToolInputGuardrailResult], + accepted_output_results: list[ToolOutputGuardrailResult], ) -> None: """Carry a turn's tool guardrail results onto the streamed result. @@ -820,6 +819,9 @@ def _accumulate_tool_guardrail_results( streamed_result.tool_output_guardrail_results = ( streamed_result.tool_output_guardrail_results + turn_result.tool_output_guardrail_results ) + if isinstance(turn_result.next_step, NextStepRunAgain | NextStepHandoff): + accepted_input_results.extend(turn_result.tool_input_guardrail_results) + accepted_output_results.extend(turn_result.tool_output_guardrail_results) async def _finalize_streamed_interruption( @@ -970,10 +972,26 @@ def _sync_conversation_tracking_from_tracker() -> None: current_agent = run_state._current_agent else: current_agent = starting_agent + _validate_resumed_session_output_guardrail_safety( + agent=current_agent, + run_config=run_config, + session=session, + run_state=run_state if is_resumed_state else None, + ) + if run_state is not None and session is None: + streamed_result._current_turn_persisted_item_count = ( + run_state._current_turn_persisted_item_count + ) if run_state is not None: current_turn = run_state._current_turn else: current_turn = 0 + accepted_tool_input_guardrail_results = ( + list(run_state._tool_input_guardrail_results) if run_state is not None else [] + ) + accepted_tool_output_guardrail_results = ( + list(run_state._tool_output_guardrail_results) if run_state is not None else [] + ) should_run_agent_start_hooks = True tool_use_tracker = AgentToolUseTracker() if run_state is not None: @@ -1146,6 +1164,13 @@ async def _save_max_turns_items( try: while True: + validate_output_guardrails_with_server_managed_conversation( + current_agent, + run_config, + conversation_id=conversation_id, + previous_response_id=previous_response_id, + auto_previous_response_id=auto_previous_response_id, + ) all_input_guardrails = ( starting_agent.input_guardrails + (run_config.input_guardrails or []) if current_turn == 0 and not is_resumed_state @@ -1227,6 +1252,25 @@ async def _save_max_turns_items( raise UserError("No processed response found in previous state") last_model_response = run_state._model_responses[-1] + resumed_response_boundary = _current_response_boundary( + (), + run_state._last_processed_response, + run_state, + ) + blocked_output_owner_starts = _BlockedOutputOwnerStarts( + run_state_generated_items=resumed_response_boundary.generated_start, + run_state_session_items=resumed_response_boundary.session_start, + run_state_model_responses=len(run_state._model_responses) - 1, + run_state_tool_output_guardrail_results=len( + run_state._tool_output_guardrail_results + ), + streamed_new_items=resumed_response_boundary.session_start, + streamed_model_input_items=resumed_response_boundary.generated_start, + streamed_raw_responses=len(streamed_result.raw_responses) - 1, + streamed_tool_output_guardrail_results=len( + streamed_result.tool_output_guardrail_results + ), + ) turn_result = await resolve_interrupted_turn( bindings=current_bindings, @@ -1299,13 +1343,25 @@ async def _save_max_turns_items( # but skips a resumed turn that loops back to the model, so a guardrail that # re-runs for the same tool call on resume is not counted twice. if not isinstance(turn_result.next_step, NextStepRunAgain): - _accumulate_tool_guardrail_results(streamed_result, turn_result) + _accumulate_tool_guardrail_results( + streamed_result, + turn_result, + accepted_input_results=accepted_tool_input_guardrail_results, + accepted_output_results=accepted_tool_output_guardrail_results, + ) if isinstance(turn_result.next_step, NextStepInterruption): await _finalize_streamed_interruption( streamed_result=streamed_result, save_items=_save_resumed_items, - items=list(turn_session_items), + items=( + [] + if _should_defer_interrupted_session_items( + current_agent, + run_config, + ) + else list(turn_session_items) + ), response_id=turn_result.model_response.response_id, store_setting=store_setting, interruptions=approvals_from_step(turn_result.next_step), @@ -1346,10 +1402,18 @@ async def _save_max_turns_items( context_wrapper=context_wrapper, save_items=_save_resumed_items, items=list(turn_session_items), + model_response=turn_result.model_response, + processed_response=( + turn_result.processed_response + if turn_result.processed_response is not None + else run_state._last_processed_response + ), + owner_starts=blocked_output_owner_starts, response_id=turn_result.model_response.response_id, store_setting=store_setting, - persist_before_output_guardrails=True, ) + if streamed_result._stored_exception is not None: + break run_state._current_step = None break @@ -1536,11 +1600,36 @@ def _record_max_turns_handler_output( context_wrapper=context_wrapper, save_items=_save_max_turns_items, items=[synthesized_item] if include_in_history else [], + model_response=None, + processed_response=None, + owner_starts=_BlockedOutputOwnerStarts( + run_state_generated_items=( + len(run_state._generated_items) if run_state is not None else None + ), + run_state_session_items=( + len(run_state._session_items) if run_state is not None else None + ), + run_state_model_responses=( + len(run_state._model_responses) if run_state is not None else None + ), + run_state_tool_output_guardrail_results=( + len(run_state._tool_output_guardrail_results) + if run_state is not None + else None + ), + streamed_new_items=len(streamed_result.new_items), + streamed_model_input_items=len(streamed_result._model_input_items), + streamed_raw_responses=len(streamed_result.raw_responses), + streamed_tool_output_guardrail_results=len( + streamed_result.tool_output_guardrail_results + ), + ), response_id=None, store_setting=store_setting, - persist_before_output_guardrails=False, on_persisted_after_guardrails=_record_max_turns_handler_output, ) + if streamed_result._stored_exception is not None: + break streamed_result._max_turns_handled = True streamed_result.current_turn = max_turns if run_state is not None and not is_resumed_state: @@ -1588,6 +1677,39 @@ def _record_max_turns_handler_output( ) ) try: + if run_state is not None and _has_output_guardrails(current_agent, run_config): + _synchronize_accepted_run_state( + run_state, + generated_items=streamed_result._model_input_items, + session_items=streamed_result.new_items, + model_responses=streamed_result.raw_responses, + tool_input_guardrail_results=accepted_tool_input_guardrail_results, + tool_output_guardrail_results=accepted_tool_output_guardrail_results, + current_turn=current_turn, + ) + + blocked_output_owner_starts = _BlockedOutputOwnerStarts( + run_state_generated_items=( + len(run_state._generated_items) if run_state is not None else None + ), + run_state_session_items=( + len(run_state._session_items) if run_state is not None else None + ), + run_state_model_responses=( + len(run_state._model_responses) if run_state is not None else None + ), + run_state_tool_output_guardrail_results=( + len(run_state._tool_output_guardrail_results) + if run_state is not None + else None + ), + streamed_new_items=len(streamed_result.new_items), + streamed_model_input_items=len(streamed_result._model_input_items), + streamed_raw_responses=len(streamed_result.raw_responses), + streamed_tool_output_guardrail_results=len( + streamed_result.tool_output_guardrail_results + ), + ) logger.debug( "Starting turn %s, current_agent=%s", current_turn, @@ -1660,7 +1782,12 @@ def _record_max_turns_handler_output( streamed_result.raw_responses = streamed_result.raw_responses + [ turn_result.model_response ] - _accumulate_tool_guardrail_results(streamed_result, turn_result) + _accumulate_tool_guardrail_results( + streamed_result, + turn_result, + accepted_input_results=accepted_tool_input_guardrail_results, + accepted_output_results=accepted_tool_output_guardrail_results, + ) input_before_turn_rewrite = streamed_result.input streamed_result.input = turn_result.original_input if isinstance(turn_result.next_step, NextStepHandoff): @@ -1738,10 +1865,14 @@ def _record_max_turns_handler_output( context_wrapper=context_wrapper, save_items=_save_stream_items_with_count, items=turn_session_items, + model_response=turn_result.model_response, + processed_response=turn_result.processed_response, + owner_starts=blocked_output_owner_starts, response_id=turn_result.model_response.response_id, store_setting=store_setting, - persist_before_output_guardrails=False, ) + if streamed_result._stored_exception is not None: + break if run_state is not None: run_state._current_step = None break @@ -1763,7 +1894,14 @@ def _record_max_turns_handler_output( await _finalize_streamed_interruption( streamed_result=streamed_result, save_items=_save_stream_items_with_count, - items=turn_session_items, + items=( + [] + if _should_defer_interrupted_session_items( + current_agent, + run_config, + ) + else turn_session_items + ), response_id=turn_result.model_response.response_id, store_setting=store_setting, interruptions=approvals_from_step(turn_result.next_step), diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index 8f33883012..f822fefea0 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -14,10 +14,12 @@ import pytest from openai import APIConnectionError, BadRequestError, NotFoundError from openai.types.responses import ResponseFunctionToolCall +from openai.types.responses.response_function_tool_call import CallerDirect, CallerProgram from openai.types.responses.response_output_item import McpApprovalRequest from openai.types.responses.response_output_text import AnnotationFileCitation, ResponseOutputText from openai.types.responses.response_reasoning_item import ResponseReasoningItem, Summary from openai.types.responses.tool_param import Mcp +from pydantic import BaseModel from typing_extensions import TypedDict import agents._debug as _debug @@ -29,6 +31,7 @@ HandoffInputData, InputGuardrail, InputGuardrailTripwireTriggered, + MaxTurnsExceeded, ModelBehaviorError, ModelRetryAdvice, ModelRetrySettings, @@ -46,12 +49,14 @@ ToolGuardrailFunctionOutput, ToolInputGuardrailData, ToolNameCollisionPolicy, + ToolOutputGuardrailData, ToolTimeoutError, UserError, handoff, retry_policies, tool_input_guardrail, tool_namespace, + tool_output_guardrail, ) from agents._tool_identity import resolve_tool_name_collisions from agents.agent import ToolsToFinalOutputResult @@ -69,8 +74,10 @@ from agents.lifecycle import RunHooks from agents.memory import SessionSettings from agents.models.fake_id import FAKE_RESPONSES_ID +from agents.result import RunResultStreaming from agents.run import AgentRunner, get_default_agent_runner, set_default_agent_runner from agents.run_config import _default_trace_include_sensitive_data +from agents.run_internal import blocked_output, run_loop from agents.run_internal.agent_bindings import bind_public_agent from agents.run_internal.agent_runner_helpers import build_resumed_stream_debug_extra from agents.run_internal.items import ( @@ -124,6 +131,835 @@ def to_input_item(self) -> dict[str, Any]: return self._payload +@pytest.mark.parametrize("arguments", ["{}", ""], ids=["json-object", "empty"]) +def test_blocked_function_batch_is_rebuilt_from_allowlisted_fields(arguments: str) -> None: + agent = Agent(name="test") + call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "commit_tool", + "arguments": arguments, + "call_id": "call-commit", + "provider_data": {"secret": "call-secret"}, + }, + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + "provider_data": {"secret": "output-secret"}, + }, + output="sdk-secret", + custom_data={"secret": "custom-secret"}, + ) + + retained = run_loop._retained_items_for_blocked_output([call, output]) + + assert [item.type for item in retained] == ["tool_call_item", "tool_call_output_item"] + retained_call = cast(ToolCallItem, retained[0]) + retained_output = cast(ToolCallOutputItem, retained[1]) + assert "provider_data" not in cast(dict[str, Any], retained_call.raw_item) + assert cast(dict[str, Any], retained_call.raw_item)["arguments"] == arguments + assert cast(dict[str, Any], retained_output.raw_item) == { + "type": "function_call_output", + "call_id": "call-commit", + "output": run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + } + assert retained_output.output == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + assert retained_output.custom_data is None + + +def test_blocked_function_batch_accepts_exact_typed_direct_caller() -> None: + agent = Agent(name="test") + call = ToolCallItem( + agent=agent, + raw_item=ResponseFunctionToolCall( + type="function_call", + name="commit_tool", + arguments="{}", + call_id="call-commit", + caller=CallerDirect(type="direct"), + ), + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + + retained = run_loop._retained_items_for_blocked_output([call, output]) + + assert len(retained) == 2 + retained_call = cast(ToolCallItem, retained[0]) + assert cast(dict[str, Any], retained_call.raw_item)["caller"] == {"type": "direct"} + + +def test_blocked_function_batch_ignores_hash_collision_key_hooks() -> None: + equality_calls: list[Any] = [] + + class HashCollisionKey: + def __init__(self, field: str) -> None: + self.field = field + + def __hash__(self) -> int: + return hash(self.field) + + def __eq__(self, other: object) -> bool: + equality_calls.append(other) + return False + + caller: dict[Any, Any] = { + HashCollisionKey("type"): "caller-secret", + "type": "direct", + } + raw_call: dict[Any, Any] = { + HashCollisionKey("type"): "type-secret", + HashCollisionKey("name"): "name-secret", + HashCollisionKey("arguments"): "arguments-secret", + HashCollisionKey("call_id"): "call-id-secret", + HashCollisionKey("id"): "id-secret", + HashCollisionKey("namespace"): "namespace-secret", + HashCollisionKey("status"): "status-secret", + HashCollisionKey("caller"): "caller-secret", + "type": "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + "caller": caller, + } + equality_calls.clear() + agent = Agent(name="test") + call = ToolCallItem(agent=agent, raw_item=cast(Any, raw_call)) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + + retained = run_loop._retained_items_for_blocked_output([call, output]) + + assert len(retained) == 2 + assert equality_calls == [] + retained_call = cast(ToolCallItem, retained[0]) + assert cast(dict[str, Any], retained_call.raw_item)["caller"] == {"type": "direct"} + + +@pytest.mark.parametrize("discriminator", ["call", "output", "caller"]) +def test_blocked_function_batch_rejects_equality_impostor_discriminators_without_hooks( + discriminator: str, +) -> None: + equality_calls: list[Any] = [] + + class EqualityImpostor: + def __eq__(self, other: object) -> bool: + equality_calls.append(other) + return True + + raw_call: dict[str, Any] = { + "type": EqualityImpostor() if discriminator == "call" else "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + } + if discriminator == "caller": + raw_call["caller"] = {"type": EqualityImpostor()} + agent = Agent(name="test") + call = ToolCallItem(agent=agent, raw_item=cast(Any, raw_call)) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": (EqualityImpostor() if discriminator == "output" else "function_call_output"), + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + + assert run_loop._retained_items_for_blocked_output([call, output]) == [] + assert equality_calls == [] + + +def test_blocked_function_batch_rejects_non_direct_typed_callers() -> None: + class GenericCaller(BaseModel): + type: str + + agent = Agent(name="test") + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + for caller in ( + CallerProgram(type="program", caller_id="program-call"), + GenericCaller(type="direct"), + ): + call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + "caller": caller, + }, + ) + + assert run_loop._retained_items_for_blocked_output([call, output]) == [] + + +def test_blocked_unknown_tool_variant_discards_the_complete_response() -> None: + agent = Agent(name="test") + call = ToolCallItem( + agent=agent, + raw_item={"type": "custom_tool_call", "call_id": "call-custom", "secret": "call"}, + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "custom_tool_call_output", + "call_id": "call-custom", + "output": "raw-secret", + }, + output="sdk-secret", + custom_data={"secret": "custom-secret"}, + ) + + assert run_loop._retained_items_for_blocked_output([call, output]) == [] + + +def test_blocked_reasoning_item_discards_the_complete_response() -> None: + agent = Agent(name="test") + reasoning = ReasoningItem( + agent=agent, + raw_item=ResponseReasoningItem( + id="reasoning-id", + type="reasoning", + summary=[], + encrypted_content="reasoning-secret", + ), + ) + call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + }, + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + + assert run_loop._retained_items_for_blocked_output([reasoning, call, output]) == [] + + +def test_blocked_snapshot_preserves_accepted_prefix_with_reused_provider_id() -> None: + agent = Agent(name="test") + prior_call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "prior_tool", + "arguments": "{}", + "call_id": "reused-call-id", + }, + ) + prior_output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "reused-call-id", + "output": "accepted-prior-output", + }, + output="accepted-prior-output", + ) + current_call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "current_tool", + "arguments": "{}", + "call_id": "reused-call-id", + }, + ) + current_output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "reused-call-id", + "output": "rejected-current-output", + }, + output="rejected-current-output", + ) + prior_response = ModelResponse( + output=[cast(Any, prior_call.raw_item)], + usage=Usage(), + response_id="prior-response", + ) + current_response = ModelResponse( + output=[cast(Any, current_call.raw_item)], + usage=Usage(), + response_id="current-response", + ) + state = make_run_state( + agent, + context=make_context_wrapper(), + original_input="test", + max_turns=2, + ) + state._generated_items = [prior_call, prior_output, current_call, current_output] + state._session_items = [prior_call, prior_output, current_call, current_output] + state._model_responses = [prior_response, current_response] + + retained = run_loop._retained_items_for_blocked_response( + [current_call, current_output], + current_response, + run_state=state, + owner_starts=run_loop._BlockedOutputOwnerStarts( + run_state_generated_items=2, + run_state_session_items=2, + run_state_model_responses=1, + run_state_tool_output_guardrail_results=0, + ), + ) + + assert state._generated_items[:2] == [prior_call, prior_output] + assert state._generated_items[0] is prior_call + assert state._generated_items[1] is prior_output + assert state._session_items[:2] == [prior_call, prior_output] + assert state._model_responses[0] is prior_response + assert retained == state._generated_items[2:] + assert cast(ToolCallOutputItem, retained[1]).output == ( + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + ) + assert prior_output.output == "accepted-prior-output" + + +def test_blocked_snapshot_cancellation_severs_replay_graph_and_propagates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + agent = Agent(name="test") + call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + }, + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + custom_data={"secret": "custom-secret"}, + ) + response = ModelResponse( + output=[cast(Any, call.raw_item)], + usage=Usage(), + response_id="response-id", + ) + state = make_run_state( + agent, + context=make_context_wrapper(), + original_input="test", + max_turns=1, + ) + state._generated_items = [call, output] + state._session_items = [call, output] + state._model_responses = [response] + cancellation = asyncio.CancelledError("original cancellation") + + def cancel_preparation(_raw_item: Any) -> dict[str, Any]: + raise cancellation + + monkeypatch.setattr(blocked_output, "blocked_function_output_payload", cancel_preparation) + + with pytest.raises(asyncio.CancelledError) as exc_info: + run_loop._retained_items_for_blocked_response( + [call, output], + response, + run_state=state, + ) + + assert exc_info.value is cancellation + assert state._generated_items == [] + assert state._session_items == [] + assert state._model_responses == [] + + +@pytest.mark.parametrize("fail_after_first_swap", [False, True]) +def test_blocked_snapshot_application_baseexception_severs_every_owner( + monkeypatch: pytest.MonkeyPatch, + fail_after_first_swap: bool, +) -> None: + agent = Agent(name="test") + prior_call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "prior_tool", + "arguments": "{}", + "call_id": "prior-call", + }, + ) + prior_output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "prior-call", + "output": "accepted-prior-output", + }, + output="accepted-prior-output", + ) + call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + }, + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + prior_response = ModelResponse( + output=[cast(Any, prior_call.raw_item)], + usage=Usage(), + response_id="prior-response", + ) + response = ModelResponse( + output=[cast(Any, call.raw_item)], + usage=Usage(), + response_id="response-id", + ) + state = make_run_state( + agent, + context=make_context_wrapper(), + original_input="test", + max_turns=1, + ) + prior_guardrail_result = cast(Any, object()) + current_guardrail_result = cast(Any, object()) + state._generated_items = [prior_call, prior_output, call, output] + state._session_items = [prior_call, prior_output, call, output] + state._model_responses = [prior_response, response] + state._tool_output_guardrail_results = [prior_guardrail_result, current_guardrail_result] + streamed_result = RunResultStreaming( + input="test", + new_items=[prior_call, prior_output, call, output], + raw_responses=[prior_response, response], + final_output=None, + input_guardrail_results=[], + output_guardrail_results=[], + tool_input_guardrail_results=[], + tool_output_guardrail_results=[prior_guardrail_result, current_guardrail_result], + context_wrapper=make_context_wrapper(), + current_agent=agent, + current_turn=2, + max_turns=2, + _current_agent_output_schema=None, + trace=None, + ) + streamed_result._model_input_items = [prior_call, prior_output, call, output] + streamed_result._state = state + application_error = KeyboardInterrupt("application failed") + + def fail_application(plan: Any) -> None: + if fail_after_first_swap: + owner, field, value = plan.assignments[0] + object.__setattr__(owner, field, value) + raise application_error + + monkeypatch.setattr(blocked_output, "_apply_blocked_output_owner_plan", fail_application) + + with pytest.raises(KeyboardInterrupt) as exc_info: + run_loop._retained_items_for_blocked_response( + [call, output], + response, + run_state=state, + streamed_result=streamed_result, + owner_starts=run_loop._BlockedOutputOwnerStarts( + run_state_generated_items=2, + run_state_session_items=2, + run_state_model_responses=1, + run_state_tool_output_guardrail_results=1, + streamed_new_items=2, + streamed_model_input_items=2, + streamed_raw_responses=1, + streamed_tool_output_guardrail_results=1, + ), + ) + + assert exc_info.value is application_error + assert state._generated_items == [prior_call, prior_output] + assert state._session_items == [prior_call, prior_output] + assert state._model_responses == [prior_response] + assert state._tool_output_guardrail_results == [prior_guardrail_result] + assert streamed_result.new_items == [prior_call, prior_output] + assert streamed_result._model_input_items == [prior_call, prior_output] + assert streamed_result.raw_responses == [prior_response] + assert streamed_result.tool_output_guardrail_results == [prior_guardrail_result] + + +def test_blocked_snapshot_application_exception_becomes_fixed_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + agent = Agent(name="test") + call = ToolCallItem( + agent=agent, + raw_item={ + "type": "function_call", + "name": "commit_tool", + "arguments": "{}", + "call_id": "call-commit", + }, + ) + output = ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call-commit", + "output": "raw-secret", + }, + output="sdk-secret", + ) + state = make_run_state( + agent, + context=make_context_wrapper(), + original_input="test", + max_turns=1, + ) + state._generated_items = [call, output] + state._session_items = [call, output] + + def fail_application(_plan: Any) -> None: + raise ValueError("application-secret") + + monkeypatch.setattr(blocked_output, "_apply_blocked_output_owner_plan", fail_application) + + with pytest.raises(RuntimeError) as exc_info: + run_loop._retained_items_for_blocked_response( + [call, output], + None, + run_state=state, + ) + + assert "application-secret" not in str(exc_info.value) + assert state._generated_items == [] + assert state._session_items == [] + + +@pytest.mark.asyncio +async def test_non_streamed_trip_preserves_prior_run_state_side_effect( + monkeypatch: pytest.MonkeyPatch, +) -> None: + side_effects: list[str] = [] + memory_items: list[RunItem] | None = None + + @function_tool(name_override="accepted_tool") + def accepted_tool() -> str: + side_effects.append("accepted") + return "accepted-output" + + @function_tool(name_override="terminal_tool") + def terminal_tool() -> str: + side_effects.append("terminal") + return "rejected-output" + + model = ScriptedModel( + steps=[ + [get_function_tool_call("accepted_tool", "{}", call_id="accepted-call")], + [get_text_message("accepted-final")], + ] + ) + agent = Agent(name="test", model=model, tools=[accepted_tool, terminal_tool]) + first = await Runner.run(agent, "run accepted tool", max_turns=5) + state = first.to_state() + prior_generated = list(state._generated_items) + prior_session = list(state._session_items) + prior_responses = list(state._model_responses) + + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + agent.tool_use_behavior = {"stop_at_tool_names": ["terminal_tool"]} + agent.output_guardrails = [OutputGuardrail(guardrail_function=reject_output)] + + async def capture_memory_payload( + _runtime: Any, + *, + input: Any, + new_items: list[Any], + final_output: object, + interruptions: list[Any], + terminal_metadata: Any, + ) -> None: + del input, final_output, interruptions, terminal_metadata + nonlocal memory_items + memory_items = cast(list[RunItem], new_items) + + monkeypatch.setattr( + "agents.run.SandboxRuntime.enqueue_memory_payload", + capture_memory_payload, + ) + model.enqueue( + [ + ResponseReasoningItem( + id="reasoning-current", + type="reasoning", + summary=[Summary(text="calling terminal tool", type="summary_text")], + ), + get_function_tool_call("terminal_tool", "{}", call_id="current-call"), + ] + ) + + with pytest.raises(OutputGuardrailTripwireTriggered): + await Runner.run(agent, state) + + assert side_effects == ["accepted", "terminal"] + assert state._generated_items == prior_generated + assert state._session_items == prior_session + assert state._model_responses == prior_responses + serialized_state = json.dumps(state.to_json()) + assert "accepted-output" in serialized_state + assert "rejected-output" not in serialized_state + assert "reasoning-current" not in serialized_state + assert memory_items is not None + serialized_memory_items = json.dumps([item.to_input_item() for item in memory_items]) + assert "accepted-output" in serialized_memory_items + assert "rejected-output" not in serialized_memory_items + assert "reasoning-current" not in serialized_memory_items + + +@pytest.mark.parametrize("streamed", [False, True], ids=["non-streamed", "streamed"]) +@pytest.mark.parametrize("handoff_turn", [False, True], ids=["run-again", "handoff"]) +@pytest.mark.asyncio +async def test_resumed_trip_preserves_accepted_turns_and_turn_budget( + streamed: bool, + handoff_turn: bool, +) -> None: + side_effects: list[str] = [] + + @tool_input_guardrail + def record_accepted_input(_data: ToolInputGuardrailData) -> ToolGuardrailFunctionOutput: + return ToolGuardrailFunctionOutput.allow(output_info="accepted-input-audit") + + @tool_output_guardrail + def record_accepted_output(_data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput: + return ToolGuardrailFunctionOutput.allow(output_info="accepted-output-audit") + + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + side_effects.append("approved") + return "approved-output" + + @function_tool( + name_override="accepted_tool", + tool_input_guardrails=[record_accepted_input], + tool_output_guardrails=[record_accepted_output], + ) + def accepted_tool() -> str: + side_effects.append("accepted") + return "accepted-output" + + @function_tool(name_override="terminal_tool") + def terminal_tool() -> str: + side_effects.append("terminal") + return "rejected-secret" + + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + model = ScriptedModel() + target = Agent( + name="target", + model=model, + tools=[terminal_tool], + tool_use_behavior={"stop_at_tool_names": ["terminal_tool"]}, + output_guardrails=[OutputGuardrail(guardrail_function=reject_output)], + ) + agent = Agent( + name="source", + model=model, + tools=[approval_tool, accepted_tool, terminal_tool], + handoffs=[target] if handoff_turn else [], + tool_use_behavior={"stop_at_tool_names": ["terminal_tool"]}, + output_guardrails=( + [] if handoff_turn else [OutputGuardrail(guardrail_function=reject_output)] + ), + ) + accepted_response = [get_function_tool_call("accepted_tool", "{}", call_id="accepted-call")] + if handoff_turn: + accepted_response.append(get_handoff_tool_call(target)) + model.extend( + [ + [get_function_tool_call("approval_tool", "{}", call_id="approved-call")], + accepted_response, + [get_function_tool_call("terminal_tool", "{}", call_id="terminal-call")], + ] + ) + + interrupted = await Runner.run(agent, "run approved tools", max_turns=3) + state = interrupted.to_state() + state.approve(interrupted.interruptions[0]) + + with pytest.raises(OutputGuardrailTripwireTriggered): + if streamed: + result = Runner.run_streamed(agent, state) + async for _ in result.stream_events(): + pass + else: + await Runner.run(agent, state) + + assert side_effects == ["approved", "accepted", "terminal"] + assert state._current_turn == 3 + assert len(state._model_responses) == 3 + assert [result.output.output_info for result in state._tool_input_guardrail_results] == [ + "accepted-input-audit" + ] + assert [result.output.output_info for result in state._tool_output_guardrail_results] == [ + "accepted-output-audit" + ] + for items in (state._generated_items, state._session_items): + outputs = [item for item in items if isinstance(item, ToolCallOutputItem)] + assert [item.output for item in outputs] == [ + "approved-output", + "accepted-output", + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT, + ] + + serialized_state = json.dumps(state.to_json()) + assert "approved-output" in serialized_state + assert "accepted-output" in serialized_state + assert "accepted-input-audit" in serialized_state + assert "accepted-output-audit" in serialized_state + assert "rejected-secret" not in serialized_state + + with pytest.raises(MaxTurnsExceeded): + await Runner.run(agent, state) + assert side_effects == ["approved", "accepted", "terminal"] + + +@pytest.mark.asyncio +async def test_non_streamed_trip_uses_safe_items_for_sandbox_memory_after_session_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + accepted_output = "accepted-tool-output" + tool_output_secret = "sandbox-memory-tool-output-secret" + persistence_secret = "sandbox-memory-session-failure-secret" + memory_items: list[RunItem] | None = None + + @function_tool(name_override="accepted_tool") + def accepted_tool() -> str: + return accepted_output + + @function_tool(name_override="terminal_tool") + def terminal_tool() -> str: + return tool_output_secret + + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + class FailingBlockedSession(SimpleListSession): + async def add_items(self, items: list[Any]) -> None: + if any( + type(item) is dict + and item.get("type") == "function_call_output" + and item.get("call_id") == "terminal-call" + for item in items + ): + raise LookupError(persistence_secret) + await super().add_items(items) + + async def capture_memory_payload( + _runtime: Any, + *, + input: Any, + new_items: list[Any], + final_output: object, + interruptions: list[Any], + terminal_metadata: Any, + ) -> None: + del input, final_output, interruptions, terminal_metadata + nonlocal memory_items + memory_items = cast(list[RunItem], new_items) + + monkeypatch.setattr( + "agents.run.SandboxRuntime.enqueue_memory_payload", + capture_memory_payload, + ) + agent = Agent( + name="test", + model=ScriptedModel( + steps=[ + [get_function_tool_call("accepted_tool", "{}", call_id="accepted-call")], + [get_function_tool_call("terminal_tool", "{}", call_id="terminal-call")], + ] + ), + tools=[accepted_tool, terminal_tool], + tool_use_behavior={"stop_at_tool_names": ["terminal_tool"]}, + output_guardrails=[OutputGuardrail(guardrail_function=reject_output)], + ) + + with pytest.raises(UserError, match="Error details are redacted") as exc_info: + await Runner.run(agent, "run terminal tool", session=FailingBlockedSession()) + + assert exc_info.value.run_data is None + assert persistence_secret not in str(exc_info.value) + assert memory_items is not None + serialized_memory_items = json.dumps([item.to_input_item() for item in memory_items]) + assert accepted_output in serialized_memory_items + assert tool_output_secret not in serialized_memory_items + assert persistence_secret not in serialized_memory_items + assert run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT in serialized_memory_items + + async def run_execute_approved_tools( agent: Agent[Any], approval_item: ToolApprovalItem, @@ -4473,8 +5309,9 @@ def guardrail_function( ] == ["user"] +@pytest.mark.parametrize("streamed", [False, True]) @pytest.mark.asyncio -async def test_output_guardrail_error_preserves_final_output_in_session() -> None: +async def test_output_guardrail_error_preserves_final_output_in_session(streamed: bool) -> None: def guardrail_function( _context: RunContextWrapper[Any], _agent: Agent[Any], _agent_output: Any ) -> GuardrailFunctionOutput: @@ -4490,7 +5327,12 @@ def guardrail_function( ) with pytest.raises(RuntimeError, match="guardrail failed"): - await Runner.run(agent, input="user_message", session=session) + if streamed: + result = Runner.run_streamed(agent, input="user_message", session=session) + async for _ in result.stream_events(): + pass + else: + await Runner.run(agent, input="user_message", session=session) items = await session.get_items() assert [ @@ -4626,7 +5468,7 @@ def commit_tool() -> str: result = await Runner.run(agent, state, session=session) assert result.final_output == "committed-result" - assert state._current_turn_persisted_item_count == 4 + assert state._current_turn_persisted_item_count == 2 items = await session.get_items() assert [ ( @@ -4641,6 +5483,11 @@ def commit_tool() -> str: ("function_call", "call-second"), ("function_call_output", "call-second"), ] + assert cast(dict[str, Any], items[-1]).get("output") == ( + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT if tripwire_triggered else "committed-result" + ) + if tripwire_triggered: + assert "committed-result" not in json.dumps(items[-2:]) @pytest.mark.asyncio diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index d962a1b393..e5c2a03fd8 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -32,6 +32,7 @@ ModelBehaviorError, ModelRetrySettings, ModelSettings, + OpenAIChatCompletionsModel, OpenAIResponsesWSModel, OutputGuardrail, OutputGuardrailTripwireTriggered, @@ -41,6 +42,7 @@ ToolGuardrailFunctionOutput, ToolInputGuardrailData, ToolOutputGuardrailData, + ToolsToFinalOutputResult, UserError, function_tool, handoff, @@ -52,10 +54,15 @@ from agents.run import RunConfig from agents.run_internal import run_loop from agents.run_internal.run_loop import QueueCompleteSentinel +from agents.run_state import RunState from agents.stream_events import AgentUpdatedStreamEvent, RawResponsesStreamEvent, StreamEvent from agents.testing import ModelStep, ScriptedModel from agents.tool import FunctionTool -from agents.tool_guardrails import tool_input_guardrail, tool_output_guardrail +from agents.tool_guardrails import ( + ToolOutputGuardrailResult, + tool_input_guardrail, + tool_output_guardrail, +) from agents.usage import Usage, _attach_raw_usage_snapshot from tests.model_test_helpers import get_response_obj @@ -2237,15 +2244,232 @@ async def test_tool() -> str: @pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) -@pytest.mark.parametrize("tripwire", [False, True], ids=["passes", "trips"]) @pytest.mark.asyncio -async def test_resumed_approved_tool_final_persists_call_output_before_output_guardrails( +async def test_run_llm_again_approval_persists_completed_sibling(mode: str) -> None: + side_effects: list[str] = [] + + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + side_effects.append("approved") + return "approved-output" + + @function_tool(name_override="sibling_tool") + def sibling_tool() -> str: + side_effects.append("sibling") + return "sibling-output" + + model = ScriptedModel( + [ + [ + get_function_tool_call("approval_tool", "{}", call_id="call-approved"), + get_function_tool_call("sibling_tool", "{}", call_id="call-sibling"), + ], + [get_text_message("done")], + ] + ) + agent = Agent( + name="test", + model=model, + tools=[approval_tool, sibling_tool], + output_guardrails=[ + OutputGuardrail( + guardrail_function=lambda _context, _agent, _output: GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + ) + ], + ) + session = SimpleListSession() + + async def run_once(input_value: Any) -> Any: + if mode == "non_streamed": + return await Runner.run(agent, input_value, session=session) + result = Runner.run_streamed(agent, input_value, session=session) + await consume_stream(result) + return result + + first = await run_once("Use both tools") + assert len(first.interruptions) == 1 + assert side_effects == ["sibling"] + + saved_before_resume = await session.get_items() + saved_sibling_items = [ + item + for item in saved_before_resume + if isinstance(item, dict) and item.get("call_id") == "call-sibling" + ] + assert [item.get("type") for item in saved_sibling_items] == [ + "function_call", + "function_call_output", + ] + assert saved_sibling_items[1].get("output") == "sibling-output" + + state = first.to_state() + assert state._current_turn_persisted_item_count > 0 + state.approve(first.interruptions[0]) + resumed = await run_once(state) + + assert resumed.final_output == "done" + assert side_effects == ["sibling", "approved"] + saved_after_resume = await session.get_items() + for call_id in ("call-sibling", "call-approved"): + assert [ + item.get("type") + for item in saved_after_resume + if isinstance(item, dict) and item.get("call_id") == call_id + ] == ["function_call", "function_call_output"] + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.parametrize("terminal_behavior", ["first", "named", "custom"]) +@pytest.mark.asyncio +async def test_terminal_behaviors_defer_completed_approval_siblings( mode: str, - tripwire: bool, + terminal_behavior: str, +) -> None: + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + return "approved-output" + + @function_tool(name_override="sibling_tool") + def sibling_tool() -> str: + return "sibling-output" + + model = ScriptedModel( + [ + [ + get_function_tool_call("approval_tool", "{}", call_id="call-approved"), + get_function_tool_call("sibling_tool", "{}", call_id="call-sibling"), + ] + ] + ) + agent = Agent( + name="test", + model=model, + tools=[approval_tool, sibling_tool], + output_guardrails=[ + OutputGuardrail( + guardrail_function=lambda _context, _agent, _output: GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + ) + ], + ) + if terminal_behavior == "first": + agent.tool_use_behavior = "stop_on_first_tool" + elif terminal_behavior == "named": + agent.tool_use_behavior = {"stop_at_tool_names": ["approval_tool"]} + else: + agent.tool_use_behavior = lambda _context, results: ToolsToFinalOutputResult( + is_final_output=True, + final_output=results[0].output, + ) + + session = SimpleListSession() + if mode == "non_streamed": + result = await Runner.run(agent, "Use both tools", session=session) + else: + result = Runner.run_streamed(agent, "Use both tools", session=session) + await consume_stream(result) + + assert len(result.interruptions) == 1 + assert result.to_state()._current_turn_persisted_item_count == 0 + assert "sibling-output" not in json.dumps(await session.get_items()) + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.parametrize("terminal_behavior", ["first", "named", "custom"]) +@pytest.mark.asyncio +async def test_persisted_run_llm_again_checkpoint_rejects_terminal_behavior_change( + mode: str, + terminal_behavior: str, ) -> None: - guardrail_state = {"tripwire": tripwire} + side_effects: list[str] = [] @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + side_effects.append("approved") + return "approved-output" + + @function_tool(name_override="sibling_tool") + def sibling_tool() -> str: + side_effects.append("sibling") + return "sibling-output" + + model = ScriptedModel( + [ + [ + get_function_tool_call("approval_tool", "{}", call_id="call-approved"), + get_function_tool_call("sibling_tool", "{}", call_id="call-sibling"), + ] + ] + ) + agent = Agent( + name="test", + model=model, + tools=[approval_tool, sibling_tool], + output_guardrails=[ + OutputGuardrail( + guardrail_function=lambda _context, _agent, _output: GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + ) + ], + ) + session = SimpleListSession() + + if mode == "non_streamed": + first = await Runner.run(agent, "Use both tools", session=session) + else: + first = Runner.run_streamed(agent, "Use both tools", session=session) + await consume_stream(first) + + assert side_effects == ["sibling"] + state = first.to_state() + assert state._current_turn_persisted_item_count > 0 + state.approve(first.interruptions[0]) + + if terminal_behavior == "first": + agent.tool_use_behavior = "stop_on_first_tool" + elif terminal_behavior == "named": + agent.tool_use_behavior = {"stop_at_tool_names": ["approval_tool"]} + else: + agent.tool_use_behavior = lambda _context, results: ToolsToFinalOutputResult( + is_final_output=True, + final_output=results[0].output, + ) + + with pytest.raises(UserError, match="after current-turn items were persisted"): + if mode == "non_streamed": + await Runner.run(agent, state, session=session) + else: + result = Runner.run_streamed(agent, state, session=session) + await consume_stream(result) + + assert side_effects == ["sibling"] + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.parametrize("outcome", ["passes", "trips", "error"]) +@pytest.mark.asyncio +async def test_resumed_approved_tool_final_persists_complete_post_verdict_batch( + mode: str, + outcome: str, +) -> None: + guardrail_state = {"outcome": outcome} + + @tool_output_guardrail + def record_tool_output(data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput: + return ToolGuardrailFunctionOutput.allow(output_info=data.output) + + @function_tool( + name_override="approval_tool", + needs_approval=True, + tool_output_guardrails=[record_tool_output], + ) def approval_tool() -> str: return "approved-result" @@ -2254,9 +2478,11 @@ def output_guardrail( _agent: Agent[Any], _output: Any, ) -> GuardrailFunctionOutput: + if guardrail_state["outcome"] == "error": + raise RuntimeError("guardrail failed") return GuardrailFunctionOutput( output_info=None, - tripwire_triggered=guardrail_state["tripwire"], + tripwire_triggered=guardrail_state["outcome"] == "trips", ) model = ScriptedModel() @@ -2282,9 +2508,15 @@ async def run_once(input_value: Any) -> Any: state = first.to_state() state.approve(first.interruptions[0]) - if tripwire: + if outcome == "trips": with pytest.raises(OutputGuardrailTripwireTriggered): await run_once(state) + assert [result.output.output_info for result in state._tool_output_guardrail_results] == [ + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + ] + elif outcome == "error": + with pytest.raises(RuntimeError, match="guardrail failed"): + await run_once(state) else: resumed = await run_once(state) assert resumed.final_output == "approved-result" @@ -2303,10 +2535,13 @@ async def run_once(input_value: Any) -> Any: ("function_call", "call-approved"), ("function_call_output", "call-approved"), ] - assert saved_tool_items[1].get("output") == "approved-result" + expected_output = ( + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT if outcome == "trips" else "approved-result" + ) + assert saved_tool_items[1].get("output") == expected_output - if tripwire: - guardrail_state["tripwire"] = False + if outcome == "trips": + guardrail_state["outcome"] = "passes" model.enqueue([get_text_message("done")]) next_result = await run_once("Continue") assert next_result.final_output == "done" @@ -2323,13 +2558,233 @@ async def run_once(input_value: Any) -> Any: ("function_call", "call-approved"), ("function_call_output", "call-approved"), ] - assert replayed_tool_items[1].get("output") == "approved-result" + assert replayed_tool_items[1].get("output") == ( + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + ) + assert "approved-result" not in json.dumps(model_input) + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_ambiguous_serialized_approval_state_fails_before_tool_execution( + mode: str, +) -> None: + tool_calls = 0 + + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + nonlocal tool_calls + tool_calls += 1 + return "secret-result" + + def output_guardrail( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=False) + + model = ScriptedModel( + [[get_function_tool_call("approval_tool", "{}", call_id="call-approved")]] + ) + agent = Agent( + name="test", + model=model, + tools=[approval_tool], + output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)], + ) + first = await Runner.run(agent, "Use approval_tool") + state = first.to_state() + state._current_turn = 2 + state._current_turn_persisted_item_count = 1 + restored = await RunState.from_json(agent, state.to_json()) + restored.approve(restored.get_interruptions()[0]) + + with pytest.raises(UserError, match="current response boundary cannot be proven"): + if mode == "non_streamed": + await Runner.run(agent, restored, session=None) + else: + result = Runner.run_streamed(agent, restored, session=None) + await consume_stream(result) + + assert tool_calls == 0 @pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.parametrize("serialized", [False, True], ids=["live", "serialized"]) +@pytest.mark.parametrize("attach_session", [False, True], ids=["without-session", "with-session"]) +@pytest.mark.asyncio +async def test_legacy_approval_checkpoint_uses_current_session_ownership( + mode: str, + serialized: bool, + attach_session: bool, +) -> None: + side_effects: list[str] = [] + + @function_tool(name_override="approval_tool", needs_approval=True) + def approval_tool() -> str: + side_effects.append("executed") + return "approved-result" + + model = ScriptedModel( + [[get_function_tool_call("approval_tool", "{}", call_id="call-approved")]] + ) + agent = Agent( + name="test", + model=model, + tools=[approval_tool], + tool_use_behavior="stop_on_first_tool", + ) + legacy_session = SimpleListSession() + first = await Runner.run(agent, "Use approval_tool", session=legacy_session) + state = first.to_state() + assert state._current_turn_persisted_item_count > 0 + if serialized: + state = await RunState.from_json(agent, state.to_json()) + state.approve(state.get_interruptions()[0]) + agent.output_guardrails = [ + OutputGuardrail( + guardrail_function=lambda _context, _agent, _output: GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + ) + ] + session = legacy_session if attach_session else None + + if attach_session: + with pytest.raises(UserError, match="after current-turn items were persisted"): + if mode == "non_streamed": + await Runner.run(agent, state, session=session) + else: + result = Runner.run_streamed(agent, state, session=session) + await consume_stream(result) + assert side_effects == [] + return + + if mode == "non_streamed": + result = await Runner.run(agent, state, session=None) + else: + result = Runner.run_streamed(agent, state, session=None) + await consume_stream(result) + + assert result.final_output == "approved-result" + assert state._current_turn_persisted_item_count == 0 + assert side_effects == ["executed"] + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.asyncio +async def test_output_guardrails_fail_closed_with_server_managed_history(mode: str) -> None: + model = ScriptedModel([[get_text_message("unreachable")]]) + agent = Agent( + name="test", + model=model, + output_guardrails=[ + OutputGuardrail( + guardrail_function=lambda _context, _agent, _output: GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + ) + ], + ) + + with pytest.raises(UserError, match="server-managed conversation history"): + if mode == "non_streamed": + await Runner.run(agent, "hello", previous_response_id="response-id") + else: + Runner.run_streamed(agent, "hello", previous_response_id="response-id") + + assert not model.calls + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.parametrize("strict", [False, True], ids=["default", "strict"]) +@pytest.mark.parametrize("use_run_config_model", [False, True], ids=["agent-model", "run-model"]) +@pytest.mark.asyncio +async def test_chat_completions_output_guardrails_use_adapter_conversation_policy( + mode: str, + strict: bool, + use_run_config_model: bool, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + scripted_model = ScriptedModel([[get_text_message("accepted-output")]]) + chat_model = OpenAIChatCompletionsModel( + model="test", + openai_client=cast(Any, object()), + strict_feature_validation=strict, + ) + + async def get_response(*args: Any, **kwargs: Any) -> Any: + chat_model._handle_unsupported_server_managed_conversation_state( + previous_response_id=kwargs.get("previous_response_id"), + conversation_id=kwargs.get("conversation_id"), + ) + return await scripted_model.get_response(*args, **kwargs) + + async def stream_response(*args: Any, **kwargs: Any) -> AsyncIterator[Any]: + chat_model._handle_unsupported_server_managed_conversation_state( + previous_response_id=kwargs.get("previous_response_id"), + conversation_id=kwargs.get("conversation_id"), + ) + async for event in scripted_model.stream_response(*args, **kwargs): + yield event + + monkeypatch.setattr(chat_model, "get_response", get_response) + monkeypatch.setattr(chat_model, "stream_response", stream_response) + agent = Agent( + name="test", + model=ScriptedModel() if use_run_config_model else chat_model, + output_guardrails=[ + OutputGuardrail( + guardrail_function=lambda _context, _agent, _output: GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + ) + ], + ) + run_config = RunConfig(model=chat_model) if use_run_config_model else None + caplog.set_level(logging.WARNING, logger="openai.agents") + + async def run_once() -> Any: + if mode == "non_streamed": + return await Runner.run( + agent, + "hello", + previous_response_id="response-id", + run_config=run_config, + ) + result = Runner.run_streamed( + agent, + "hello", + previous_response_id="response-id", + run_config=run_config, + ) + await consume_stream(result) + return result + + if strict: + with pytest.raises(UserError, match="OpenAIChatCompletionsModel does not support"): + await run_once() + assert not scripted_model.calls + return + + assert (await run_once()).final_output == "accepted-output" + assert "Ignoring unsupported server-managed conversation state" in caplog.text + assert len(scripted_model.calls) == 1 + + +@pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) +@pytest.mark.parametrize("session_kind", ["simple", "openai_conversations"]) +@pytest.mark.parametrize("arguments", ["{}", ""], ids=["json-object", "empty"]) @pytest.mark.asyncio async def test_stop_on_first_tool_final_persists_committed_tool_items_on_tripwire( mode: str, + session_kind: str, + arguments: str, ) -> None: """A blocked final output must not discard the session record of a tool that already ran.""" @@ -2348,7 +2803,7 @@ def output_guardrail( return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) model = ScriptedModel() - model.enqueue([get_function_tool_call("commit_tool", "{}", call_id="call-committed")]) + model.enqueue([get_function_tool_call("commit_tool", arguments, call_id="call-committed")]) agent = Agent( name="test", model=model, @@ -2356,13 +2811,38 @@ def output_guardrail( tool_use_behavior="stop_on_first_tool", output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)], ) - session = SimpleListSession() + + class DummyOpenAIConversationsSession(OpenAIConversationsSession): + def __init__(self) -> None: + self.history: list[TResponseInputItem] = [] + + async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]: + return list(self.history if limit is None else self.history[-limit:]) + + async def add_items(self, items: list[TResponseInputItem]) -> None: + self.history.extend(items) + + async def pop_item(self) -> TResponseInputItem | None: + return self.history.pop() if self.history else None + + async def clear_session(self) -> None: + self.history.clear() + + session = SimpleListSession() if session_kind == "simple" else DummyOpenAIConversationsSession() + run_config = RunConfig( + session_input_callback=lambda history, new_input: [*reversed(history), *new_input] + ) with pytest.raises(OutputGuardrailTripwireTriggered): if mode == "non_streamed": - await Runner.run(agent, "Use commit_tool", session=session) + await Runner.run(agent, "Use commit_tool", session=session, run_config=run_config) else: - result = Runner.run_streamed(agent, "Use commit_tool", session=session) + result = Runner.run_streamed( + agent, + "Use commit_tool", + session=session, + run_config=run_config, + ) await consume_stream(result) assert calls == ["ran"], "the tool never ran, so the test proves nothing" @@ -2378,14 +2858,28 @@ def output_guardrail( ("function_call", "call-committed"), ("function_call_output", "call-committed"), ] + assert cast(dict[str, Any], saved_items[-1]).get("output") == ( + run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + ) + assert "committed-result" not in json.dumps(saved_items) # The next run must see the completed call instead of re-issuing the same side effect. agent.output_guardrails = [] model.enqueue([get_text_message("done")]) if mode == "non_streamed": - followup: Any = await Runner.run(agent, "Continue", session=session) + followup: Any = await Runner.run( + agent, + "Continue", + session=session, + run_config=run_config, + ) else: - followup = Runner.run_streamed(agent, "Continue", session=session) + followup = Runner.run_streamed( + agent, + "Continue", + session=session, + run_config=run_config, + ) await consume_stream(followup) assert followup.final_output == "done" assert calls == ["ran"] @@ -2397,10 +2891,17 @@ def output_guardrail( for item in model_input if isinstance(item, dict) and item.get("type") in {"function_call", "function_call_output"} ] - assert replayed == [ + assert set(replayed) == { ("function_call", "call-committed"), ("function_call_output", "call-committed"), - ] + } + replayed_output = next( + item.get("output") + for item in model_input + if isinstance(item, dict) and item.get("type") == "function_call_output" + ) + assert replayed_output == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + assert "committed-result" not in json.dumps(model_input) @pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) @@ -2450,7 +2951,10 @@ def output_guardrail( _agent: Agent[Any], _output: Any, ) -> GuardrailFunctionOutput: - return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + return GuardrailFunctionOutput( + output_info={"reason": "message rejected"}, + tripwire_triggered=True, + ) model = ScriptedModel() model.extend( @@ -2467,13 +2971,16 @@ def output_guardrail( ) session = SimpleListSession() - with pytest.raises(OutputGuardrailTripwireTriggered): + with pytest.raises(OutputGuardrailTripwireTriggered) as exc_info: if mode == "non_streamed": await Runner.run(agent, "Use commit_tool", session=session) else: result = Runner.run_streamed(agent, "Use commit_tool", session=session) await consume_stream(result) + assert exc_info.value.guardrail_result.agent_output == "should_not_be_saved" + assert exc_info.value.guardrail_result.output.output_info == {"reason": "message rejected"} + saved_items = await session.get_items() saved = [item.get("type") or item.get("role") for item in saved_items if isinstance(item, dict)] assert saved == ["user", "function_call", "function_call_output"] @@ -2549,12 +3056,7 @@ async def run_once() -> Any: async def test_failing_output_guardrail_keeps_the_whole_final_turn( mode: str, ) -> None: - """A guardrail *error* is not a tripwire: the completed final turn stays replayable. - - Only a tripwire means the output was judged undeliverable. An ordinary guardrail exception - leaves the verdict unknown, so the turn must be persisted whole, exactly as the non-streamed - path does. - """ + """A guardrail error leaves no rejection, so the completed turn remains replayable.""" @function_tool(name_override="commit_tool") def commit_tool() -> str: @@ -2772,16 +3274,11 @@ def output_guardrail( @pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) @pytest.mark.parametrize("tripwire", [False, True], ids=["passes", "trips"]) @pytest.mark.asyncio -async def test_blocked_tool_final_keeps_reasoning_context_with_the_committed_call( +async def test_blocked_tool_final_discards_reasoning_response_suffix_on_trip( mode: str, tripwire: bool, ) -> None: - """A retained tool call keeps the reasoning item it belongs to, in order. - - A reasoning model requires the reasoning item that preceded a function call to accompany that - call in the next request, so persisting the call/output pair without it leaves an unreplayable - turn. Asserted on both the session contents and the next run's model input. - """ + """A reasoning-bearing response is preserved on pass and discarded completely on trip.""" @function_tool(name_override="commit_tool") def commit_tool() -> str: @@ -2829,7 +3326,10 @@ async def run_once(input_value: Any) -> Any: saved_items = await session.get_items() saved = [item.get("type") or item.get("role") for item in saved_items if isinstance(item, dict)] - assert saved == ["user", "reasoning", "function_call", "function_call_output"] + expected_saved = ( + ["user"] if tripwire else ["user", "reasoning", "function_call", "function_call_output"] + ) + assert saved == expected_saved # The reasoning/call/output group has to reach the next request in that order. agent.output_guardrails = [] @@ -2845,22 +3345,16 @@ async def run_once(input_value: Any) -> Any: if isinstance(item, dict) and item.get("type") in {"reasoning", "function_call", "function_call_output"} ] - assert replayed == ["reasoning", "function_call", "function_call_output"] + expected_replayed = [] if tripwire else ["reasoning", "function_call", "function_call_output"] + assert replayed == expected_replayed @pytest.mark.parametrize("mode", ["non_streamed", "streamed"]) @pytest.mark.asyncio -async def test_blocked_tool_final_drops_reasoning_tied_to_the_rejected_message( +async def test_blocked_tool_final_discards_suffix_with_multiple_reasoning_groups( mode: str, ) -> None: - """Only the reasoning tied to a retained call survives; the message's reasoning goes with it. - - The turn is `reasoning_for_message -> message -> reasoning_for_call -> function_call`. A - reasoning item belongs to the next non-reasoning item, so retaining every reasoning item - whenever the turn happens to contain a tool call would leave the rejected message's reasoning - dangling in the next request. - - """ + """Any reasoning item makes the complete rejected current-response suffix unsupported.""" @function_tool(name_override="commit_tool") def commit_tool() -> str: @@ -2911,17 +3405,14 @@ async def run_once(input_value: Any) -> Any: saved_items = await session.get_items() saved = [item.get("type") or item.get("role") for item in saved_items if isinstance(item, dict)] - assert saved == ["user", "reasoning", "function_call", "function_call_output"] + assert saved == ["user"] saved_reasoning_ids = [ item.get("id") for item in saved_items if isinstance(item, dict) and item.get("id") ] - assert "rs_committed" in saved_reasoning_ids - assert "rs_rejected" not in saved_reasoning_ids, ( - "reasoning tied to the rejected message must not be persisted" - ) + assert saved_reasoning_ids == [] - # ...and the surviving group still replays in order, with no dangling reasoning item. + # The unsupported response contributes nothing to the next model request. agent.output_guardrails = [] model.enqueue([get_text_message("done")]) followup = await run_once("Continue") @@ -2935,7 +3426,94 @@ async def run_once(input_value: Any) -> Any: if isinstance(item, dict) and item.get("type") in {"reasoning", "message", "function_call", "function_call_output"} ] - assert replayed == ["reasoning", "function_call", "function_call_output"] + assert replayed == [] + + +@pytest.mark.parametrize("reasoning_suffix", [False, True], ids=["canonical", "reasoning"]) +@pytest.mark.asyncio +async def test_streamed_trip_preserves_accepted_tool_prefix( + reasoning_suffix: bool, +) -> None: + """Only the rejected current response is replaced or dropped from replay owners.""" + side_effects: list[str] = [] + + @function_tool(name_override="accepted_tool") + def accepted_tool() -> str: + side_effects.append("accepted") + return "accepted-output" + + @function_tool(name_override="terminal_tool") + def terminal_tool() -> str: + side_effects.append("terminal") + return "rejected-output" + + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + terminal_response: list[Any] = [] + if reasoning_suffix: + terminal_response.append( + ResponseReasoningItem( + id="reasoning-current", + summary=[Summary(text="calling terminal tool", type="summary_text")], + type="reasoning", + ) + ) + terminal_response.append(get_function_tool_call("terminal_tool", "{}", call_id="current-call")) + model = ScriptedModel( + steps=[ + [get_function_tool_call("accepted_tool", "{}", call_id="accepted-call")], + terminal_response, + ] + ) + agent = Agent( + name="test", + model=model, + tools=[accepted_tool, terminal_tool], + tool_use_behavior={"stop_at_tool_names": ["terminal_tool"]}, + output_guardrails=[OutputGuardrail(guardrail_function=reject_output)], + ) + + result = Runner.run_streamed(agent, "run both tools") + with pytest.raises(OutputGuardrailTripwireTriggered): + await consume_stream(result) + + assert side_effects == ["accepted", "terminal"] + + def call_ids(items: list[RunItem]) -> list[str]: + return [ + call_id + for item in items + if ( + call_id := ( + item.raw_item.get("call_id") + if isinstance(item.raw_item, dict) + else getattr(item.raw_item, "call_id", None) + ) + ) + is not None + ] + + expected_call_ids = ["accepted-call", "accepted-call"] + if not reasoning_suffix: + expected_call_ids.extend(["current-call", "current-call"]) + assert call_ids(result.new_items) == expected_call_ids + assert call_ids(result._model_input_items) == expected_call_ids + + state = result.to_state() + assert call_ids(state._generated_items) == expected_call_ids + assert call_ids(state._session_items) == expected_call_ids + serialized_state = json.dumps(state.to_json()) + assert "accepted-output" in serialized_state + assert "rejected-output" not in serialized_state + if reasoning_suffix: + assert "reasoning-current" not in serialized_state + else: + assert run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT in serialized_state @pytest.mark.asyncio @@ -3263,6 +3841,84 @@ async def test_streamed_run_reports_tool_guardrail_results(): assert result.tool_output_guardrail_results[0].output.output_info == "output-checked" +@pytest.mark.parametrize("tool_guardrail_behavior", ["allow", "reject_content"]) +@pytest.mark.asyncio +async def test_streamed_trip_replaces_current_tool_output_guardrail_results( + tool_guardrail_behavior: str, +) -> None: + """A copied terminal tool result is replaced in public and RunState guardrail results.""" + original_outputs: list[ToolGuardrailFunctionOutput] = [] + + @tool_output_guardrail + def retain_output(data: ToolOutputGuardrailData) -> ToolGuardrailFunctionOutput: + if tool_guardrail_behavior == "reject_content": + output = ToolGuardrailFunctionOutput.reject_content( + message=f"Rejected sensitive tool output: {data.output}", + output_info=data.output, + ) + else: + output = ToolGuardrailFunctionOutput.allow(output_info=data.output) + original_outputs.append(output) + return output + + @function_tool(name_override="secret_tool", tool_output_guardrails=[retain_output]) + def secret_tool() -> str: + return "blocked-secret" + + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + model = ScriptedModel() + model.enqueue([get_function_tool_call("secret_tool", "{}", call_id="call-secret")]) + agent = Agent( + name="test", + model=model, + tools=[secret_tool], + tool_use_behavior="stop_on_first_tool", + output_guardrails=[OutputGuardrail(guardrail_function=reject_output)], + ) + + result = Runner.run_streamed(agent, "run") + prior_output = ToolGuardrailFunctionOutput.allow(output_info="prior-safe") + prior_result = ToolOutputGuardrailResult(guardrail=retain_output, output=prior_output) + result.tool_output_guardrail_results.append(prior_result) + with pytest.raises(OutputGuardrailTripwireTriggered): + await consume_stream(result) + + assert len(original_outputs) == 1 + assert len(result.tool_output_guardrail_results) == 2 + assert result.tool_output_guardrail_results[0] is prior_result + assert result.tool_output_guardrail_results[0].output is prior_output + public_output = result.tool_output_guardrail_results[1].output + assert public_output is not original_outputs[0] + assert public_output.output_info == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + assert public_output.behavior["type"] == tool_guardrail_behavior + if public_output.behavior["type"] == "reject_content": + assert public_output.behavior["message"] == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + assert original_outputs[0].behavior["type"] == "reject_content" + assert "blocked-secret" in original_outputs[0].behavior["message"] + assert result._state is not None + # The caller-added public result was never owned by RunState, so only the current + # data-free result is added to that owner. + assert len(result._state._tool_output_guardrail_results) == 1 + state_output = result._state._tool_output_guardrail_results[0].output + assert state_output is public_output + assert state_output.output_info == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + assert state_output.behavior["type"] == tool_guardrail_behavior + serialized_state = result.to_state().to_json() + serialized_results = serialized_state["tool_output_guardrail_results"] + assert serialized_results[0]["output"]["behavior"]["type"] == "allow" + serialized_behavior = serialized_results[-1]["output"]["behavior"] + assert serialized_behavior["type"] == tool_guardrail_behavior + if tool_guardrail_behavior == "reject_content": + assert serialized_behavior["message"] == run_loop._OUTPUT_GUARDRAIL_BLOCKED_TOOL_OUTPUT + assert "blocked-secret" not in json.dumps(serialized_state) + + @pytest.mark.asyncio async def test_streamed_tool_guardrail_results_match_non_streamed(): """The same run reports the same tool guardrail results in both execution modes.""" diff --git a/tests/test_error_logging_redaction.py b/tests/test_error_logging_redaction.py index 6c26da31f0..1de64711df 100644 --- a/tests/test_error_logging_redaction.py +++ b/tests/test_error_logging_redaction.py @@ -1686,6 +1686,69 @@ def output_guardrail( _assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET) +@pytest.mark.parametrize("streamed", [False, True]) +@pytest.mark.asyncio +async def test_blocked_terminal_tool_session_failure_is_data_redacted( + streamed: bool, +) -> None: + tool_output_secret = "BLOCKED_TERMINAL_TOOL_OUTPUT_SECRET" + persistence_secret = "BLOCKED_TERMINAL_SESSION_FAILURE_SECRET" + + @function_tool(name_override="terminal_tool") + def terminal_tool() -> str: + return tool_output_secret + + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + class FailingBlockedSession(SimpleListSession): + async def add_items(self, items: list[Any]) -> None: + if any( + type(item) is dict and item.get("type") == "function_call_output" for item in items + ): + error = LookupError(f"session save failed: {persistence_secret}") + error.run_data = items # type: ignore[attr-defined] + raise error + await super().add_items(items) + + agent = Agent( + name="test", + model=ScriptedModel( + steps=[[get_function_tool_call("terminal_tool", "{}", call_id="terminal-call")]] + ), + tools=[terminal_tool], + tool_use_behavior="stop_on_first_tool", + output_guardrails=[OutputGuardrail(guardrail_function=reject_output)], + ) + session = FailingBlockedSession() + + if streamed: + result = Runner.run_streamed(agent, "run terminal tool", session=session) + with pytest.raises(UserError) as exc_info: + async for _ in result.stream_events(): + pass + else: + with pytest.raises(UserError) as exc_info: + await Runner.run(agent, "run terminal tool", session=session) + + error = exc_info.value + assert str(error) == "Error details are redacted." + assert error.run_data is None + assert error.__cause__ is None + assert error.__context__ is None + for secret in (tool_output_secret, persistence_secret): + assert secret not in repr(error) + _assert_secret_absent_from_agents_traceback( + error, + secret, + require_agents_frames=False, + ) + + def _persistence_failure( kind: Literal["exception", "cancelled", "direct_base", "exception_group", "group"], secret: str, diff --git a/tests/test_max_turns.py b/tests/test_max_turns.py index d002609523..02817e57cc 100644 --- a/tests/test_max_turns.py +++ b/tests/test_max_turns.py @@ -662,8 +662,10 @@ async def run_once() -> Any: with pytest.raises(RuntimeError, match="guardrail failed"): await run_once() elif outcome == "tripwire": - with pytest.raises(OutputGuardrailTripwireTriggered): + with pytest.raises(OutputGuardrailTripwireTriggered) as exc_info: await run_once() + assert exc_info.value.guardrail_result.agent_output == "fallback answer" + assert exc_info.value.guardrail_result.output.output_info == "tripwire" else: result = await run_once() assert result.final_output == "fallback answer" @@ -672,10 +674,10 @@ async def run_once() -> Any: saved_items = await session.get_items() saved_types = [str(item.get("type", item.get("role"))) for item in saved_items] - if outcome == "tripwire": - assert saved_types == ["user"] - else: + if outcome in {"pass", "error"}: assert saved_types == ["user", "message"] + else: + assert saved_types == ["user"] fallback_events = [ event @@ -688,7 +690,7 @@ async def run_once() -> Any: if streamed: assert streamed_result is not None - expected_history_count = 0 if outcome == "tripwire" else 1 + expected_history_count = 1 if outcome in {"pass", "error"} else 0 assert ( len([item for item in streamed_result.new_items if isinstance(item, MessageOutputItem)]) == expected_history_count @@ -700,6 +702,64 @@ async def run_once() -> Any: ) +@pytest.mark.asyncio +async def test_streamed_max_turns_trip_preserves_completed_tool_prefix() -> None: + def reject_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + model = ScriptedModel( + steps=[[get_function_tool_call("some_function", "{}", call_id="accepted-call")]] + ) + agent = Agent( + name="test", + model=model, + tools=[get_function_tool("some_function", "accepted-output")], + output_guardrails=[OutputGuardrail(guardrail_function=reject_output)], + ) + session = SimpleListSession() + result = Runner.run_streamed( + agent, + "run the tool", + max_turns=1, + session=session, + error_handlers={"max_turns": lambda data: "rejected fallback"}, + ) + + with pytest.raises(OutputGuardrailTripwireTriggered): + async for _ in result.stream_events(): + pass + + def call_ids(items: list[Any]) -> list[str]: + return [ + call_id + for item in items + if ( + call_id := ( + item.raw_item.get("call_id") + if isinstance(item.raw_item, dict) + else getattr(item.raw_item, "call_id", None) + ) + ) + is not None + ] + + assert call_ids(result.new_items) == ["accepted-call", "accepted-call"] + assert call_ids(result._model_input_items) == ["accepted-call", "accepted-call"] + state = result.to_state() + assert call_ids(state._generated_items) == ["accepted-call", "accepted-call"] + assert call_ids(state._session_items) == ["accepted-call", "accepted-call"] + serialized_state = json.dumps(state.to_json()) + assert "accepted-output" in serialized_state + assert "rejected fallback" in serialized_state + + saved_types = [item.get("type", item.get("role")) for item in await session.get_items()] + assert saved_types == ["user", "function_call", "function_call_output"] + + @pytest.mark.asyncio async def test_streamed_max_turns_handler_validation_failure_persists_input() -> None: agent = Agent(name="test", model=ScriptedModel(), output_type=Foo) @@ -995,6 +1055,52 @@ def output_guardrail( ] +@pytest.mark.parametrize("streamed", [False, True]) +@pytest.mark.asyncio +async def test_resumed_max_turns_trip_preserves_current_guardrail_result( + streamed: bool, +) -> None: + def output_guardrail( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=output, + tripwire_triggered=output == "fallback answer", + ) + + agent = Agent( + name="test", + model=ScriptedModel(steps=[[get_text_message("first response")]]), + output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)], + ) + first = await Runner.run(agent, "first input", max_turns=1) + state = first.to_state() + prior_result = state._output_guardrail_results[0] + + with pytest.raises(OutputGuardrailTripwireTriggered) as exc_info: + if streamed: + result = Runner.run_streamed( + agent, + state, + error_handlers={"max_turns": lambda data: "fallback answer"}, + ) + async for _ in result.stream_events(): + pass + else: + await Runner.run( + agent, + state, + error_handlers={"max_turns": lambda data: "fallback answer"}, + ) + + assert state._output_guardrail_results == [prior_result] + assert prior_result.output.output_info == "first response" + assert exc_info.value.guardrail_result.agent_output == "fallback answer" + assert exc_info.value.guardrail_result.output.output_info == "fallback answer" + + @pytest.mark.parametrize("streamed", [False, True]) @pytest.mark.asyncio async def test_resumed_max_turns_handler_preserves_checkpoint_after_continuation(