github-featured/openai--openai-agents-python
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity
Files
main
openai--openai-agents-python/tests/test_error_logging_redaction.py
T

2894 lines
100 KiB
Python
Raw Permalink Blame History

"""Error-path logging must not leak model/tool payloads when data logging is disabled.
The exception attached to a ``SpanError`` is already redacted based on the tracing
flag, but the sibling ``logger.error`` calls used to log the raw exception (and, for
tool actions, the full traceback) unconditionally. These tests lock in that those log
statements honor ``_debug.DONT_LOG_MODEL_DATA`` / ``_debug.DONT_LOG_TOOL_DATA``.
"""
from __future__ import annotations
import asyncio
import json
import logging
import pickle
import sys
import threading
import traceback
import warnings
from logging.handlers import QueueHandler
from pathlib import Path
from queue import SimpleQueue
from typing import Any, Literal, cast
from unittest.mock import patch
import httpx
import pytest
from openai import AsyncOpenAI
from pydantic import BaseModel, SkipValidation, ValidationError
import agents._debug as _debug
from agents import (
Agent,
GuardrailFunctionOutput,
InputGuardrail,
ModelBehaviorError,
ModelSettings,
ModelTracing,
OpenAIResponsesModel,
OutputGuardrail,
RunConfig,
RunContextWrapper,
RunErrorHandlerInput,
RunErrorHandlerResult,
Runner,
RunResultStreaming,
UserError,
function_tool,
handoff,
trace,
)
from agents.agent_output import AgentOutputSchema
from agents.exceptions import (
BaseExceptionGroup,
_detach_data_redacted_error_traceback,
_mark_error_data_redacted,
)
from agents.logger import (
log_model_action_debug,
log_model_action_error,
log_model_action_warning,
log_model_and_tool_action_debug,
log_model_and_tool_action_error,
log_model_and_tool_action_warning,
log_model_and_tool_data_warning,
log_tool_action_debug,
log_tool_action_error as log_shared_tool_action_error,
log_tool_action_warning,
)
from agents.realtime import RealtimeAgent, realtime_handoff
from agents.run import AgentRunner
from agents.run_internal.run_loop import _safe_redacted_persistence_error
from agents.run_internal.tool_execution import (
log_tool_action_error,
resolve_approval_rejection_message,
)
from agents.run_state import _deserialize_items
from agents.testing import ScriptedModel
from agents.tool_context import ToolContext
from agents.tracing.processor_interface import TracingProcessor
from agents.tracing.provider import SynchronousMultiTracingProcessor
from agents.tracing.spans import Span
from agents.tracing.traces import Trace
from .test_responses import get_function_tool_call, get_text_message
from .utils.simple_session import SimpleListSession
_SECRET = "super secret prompt content"
class _RecordingHandler(logging.Handler):
def __init__(self) -> None:
super().__init__()
self.records: list[logging.LogRecord] = []
def emit(self, record: logging.LogRecord) -> None:
self.records.append(record)
class _HostileException(Exception):
def __str__(self) -> str:
raise AssertionError("redacted logging inspected __str__")
def __repr__(self) -> str:
raise AssertionError("redacted logging inspected __repr__")
def __getattribute__(self, name: str):
if name in {"__class__", "__traceback__"}:
raise AssertionError(f"redacted logging inspected {name}")
return super().__getattribute__(name)
class _HostileAttributeWriteException(Exception):
def __setattr__(self, name: str, value: Any) -> None:
raise RuntimeError("redacted handling mutated the handler exception")
class _DirectBaseException(BaseException):
pass
class _HostileClassBaseException(BaseException):
@property
def __class__(self) -> type[object]:
raise RuntimeError("hostile class descriptor secret")
class _HybridCancelledError(asyncio.CancelledError, Exception):
pass
class _TruthinessException(Exception):
def __init__(self, *, truthy: bool) -> None:
super().__init__("diagnostic failure")
self.truthy = truthy
self.bool_calls = 0
def __bool__(self) -> bool:
self.bool_calls += 1
if self.truthy:
raise AssertionError("logging inspected exception truthiness")
return False
class _HostileValue:
def __str__(self) -> str:
raise AssertionError("redacted logging inspected __str__")
def __repr__(self) -> str:
raise AssertionError("redacted logging inspected __repr__")
class _FailingTracingProcessor(TracingProcessor):
def __init__(self) -> None:
self.str_calls = 0
self.lock = threading.Lock()
def __str__(self) -> str:
self.str_calls += 1
return "SECRET_TRACE_PROCESSOR_ID"
def _fail(self) -> None:
raise ValueError(_SECRET)
def on_trace_start(self, trace: Trace) -> None:
self._fail()
def on_trace_end(self, trace: Trace) -> None:
self._fail()
def on_span_start(self, span: Span[Any]) -> None:
self._fail()
def on_span_end(self, span: Span[Any]) -> None:
self._fail()
def shutdown(self) -> None:
self._fail()
def force_flush(self) -> None:
self._fail()
def _emit_shared_error_for_location(test_logger, helper) -> None:
helper(test_logger, "Fixed operational message", ValueError("failure"))
def _emit_tool_execution_error_for_location() -> None:
log_tool_action_error("Fixed operational message", ValueError("failure"))
def _emit_data_warning_for_location(test_logger) -> None:
log_model_and_tool_data_warning(
test_logger,
"Fixed operational warning",
diagnostic_message="Warning for %s",
diagnostic_args=lambda: ("diagnostic-value",),
)
def _responses_model() -> OpenAIResponsesModel:
return OpenAIResponsesModel(
model="test-model",
openai_client=AsyncOpenAI(
api_key="test",
http_client=httpx.AsyncClient(trust_env=False),
),
)
@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_error_redacts_exception_from_logs(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
model = _responses_model()
async def raise_fetch(*args, **kwargs):
raise ValueError(_SECRET)
monkeypatch.setattr(model, "_fetch_response", raise_fetch)
with patch("agents.models.openai_responses.logger") as mock_logger:
with trace(workflow_name="test"):
with pytest.raises(ValueError):
await model.get_response(
"instr",
"input",
ModelSettings(),
[],
None,
[],
ModelTracing.ENABLED,
previous_response_id=None,
)
mock_logger.error.assert_called_once()
logged = str(mock_logger.error.call_args)
assert _SECRET not in logged
assert "ValueError" not in logged
assert "Error getting response" in logged
@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_get_response_error_logs_exception_when_model_data_enabled(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
model = _responses_model()
async def raise_fetch(*args, **kwargs):
raise ValueError(_SECRET)
monkeypatch.setattr(model, "_fetch_response", raise_fetch)
with patch("agents.models.openai_responses.logger") as mock_logger:
with trace(workflow_name="test"):
with pytest.raises(ValueError):
await model.get_response(
"instr",
"input",
ModelSettings(),
[],
None,
[],
ModelTracing.ENABLED,
previous_response_id=None,
)
mock_logger.error.assert_called_once()
assert _SECRET in str(mock_logger.error.call_args)
@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
async def test_stream_response_error_redacts_exception_from_logs(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
model = _responses_model()
async def raise_fetch(*args, **kwargs):
raise ValueError(_SECRET)
monkeypatch.setattr(model, "_fetch_response", raise_fetch)
with patch("agents.models.openai_responses.logger") as mock_logger:
with trace(workflow_name="test"):
with pytest.raises(ValueError):
async for _ in model.stream_response(
"instr",
"input",
ModelSettings(),
[],
None,
[],
ModelTracing.ENABLED,
previous_response_id=None,
):
pass
mock_logger.error.assert_called_once()
logged = str(mock_logger.error.call_args)
assert _SECRET not in logged
assert "ValueError" not in logged
assert "Error streaming response" in logged
def test_log_tool_action_error_redacts_by_default(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
with patch("agents.run_internal.tool_execution.logger") as mock_logger:
log_tool_action_error("Shell executor failed", ValueError("rm -rf /secret/path"))
mock_logger.error.assert_called_once()
assert mock_logger.error.call_args.args == ("%s", "Shell executor failed")
# No traceback either, since it can embed the same sensitive data.
assert mock_logger.error.call_args.kwargs.get("exc_info") in (None, False)
@pytest.mark.parametrize(
("helper", "model_flag", "tool_flag"),
[
(log_model_action_error, True, False),
(log_model_action_debug, True, False),
(log_model_action_warning, True, False),
(log_tool_action_debug, False, True),
(log_shared_tool_action_error, False, True),
(log_tool_action_warning, False, True),
(log_model_and_tool_action_error, True, False),
(log_model_and_tool_action_error, False, True),
(log_model_and_tool_action_debug, True, False),
(log_model_and_tool_action_warning, False, True),
],
)
def test_shared_error_helpers_do_not_inspect_or_attach_redacted_exceptions(
monkeypatch,
helper,
model_flag: bool,
tool_flag: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", model_flag)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_flag)
test_logger = logging.Logger("sensitive-logging-redacted")
handler = _RecordingHandler()
test_logger.addHandler(handler)
hostile = _HostileException()
helper(test_logger, "Fixed operational message", hostile)
assert len(handler.records) == 1
record = handler.records[0]
assert record.msg == "%s"
assert record.args == ("Fixed operational message",)
assert record.exc_info is None
assert record.exc_text is None
assert hostile not in record.__dict__.values()
assert logging.Formatter().format(record) == "Fixed operational message"
def test_shared_error_helper_preserves_diagnostics_when_enabled(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
test_logger = logging.Logger("sensitive-logging-diagnostic")
handler = _RecordingHandler()
test_logger.addHandler(handler)
error = ValueError(_SECRET)
log_shared_tool_action_error(test_logger, "Tool failed", error)
record = handler.records[0]
assert isinstance(record.args, tuple)
assert error in record.args
assert record.exc_info is not None
assert record.exc_info[1] is error
assert _SECRET in logging.Formatter().format(record)
@pytest.mark.parametrize(
"helper",
[log_shared_tool_action_error, log_tool_action_warning],
)
@pytest.mark.parametrize("truthy", [False, True], ids=["falsey", "hostile_bool"])
def test_shared_error_helpers_do_not_evaluate_exception_truthiness(
monkeypatch,
helper,
truthy: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
test_logger = logging.Logger("sensitive-logging-exception-truthiness")
handler = _RecordingHandler()
test_logger.addHandler(handler)
error = _TruthinessException(truthy=truthy)
try:
raise error
except _TruthinessException:
helper(test_logger, "Tool failed", error)
record = handler.records[0]
assert error.bool_calls == 0
assert record.exc_info is not None
assert record.exc_info[0] is type(error)
assert record.exc_info[1] is error
assert record.exc_info[2] is error.__traceback__
@pytest.mark.parametrize("redacted", [True, False])
def test_shared_error_helper_conditionally_attaches_diagnostic_extra(
monkeypatch, redacted: bool
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", redacted)
test_logger = logging.Logger("sensitive-logging-diagnostic-extra")
handler = _RecordingHandler()
test_logger.addHandler(handler)
extra_calls = 0
def diagnostic_extra() -> dict[str, object]:
nonlocal extra_calls
extra_calls += 1
return {"sandbox_id": _SECRET}
log_tool_action_warning(
test_logger,
"Tool failed",
ValueError("failure"),
diagnostic_extra=diagnostic_extra,
)
record = handler.records[0]
assert extra_calls == (0 if redacted else 1)
assert ("openai_agents_diagnostic_context" in record.__dict__) is not redacted
if not redacted:
assert record.__dict__["openai_agents_diagnostic_context"] == {"sandbox_id": _SECRET}
def test_shared_error_helper_ignores_diagnostic_extra_failure(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
test_logger = logging.Logger("sensitive-logging-diagnostic-extra-failure")
handler = _RecordingHandler()
test_logger.addHandler(handler)
error = RuntimeError("original failure")
def diagnostic_extra() -> dict[str, object]:
raise AttributeError("metadata failure")
log_tool_action_warning(
test_logger,
"Tool failed",
error,
diagnostic_extra=diagnostic_extra,
)
record = handler.records[0]
assert "openai_agents_diagnostic_context" not in record.__dict__
assert record.exc_info is not None
assert record.exc_info[1] is error
assert "original failure" in logging.Formatter().format(record)
def test_shared_data_warning_does_not_inspect_or_attach_redacted_arguments(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
test_logger = logging.Logger("sensitive-logging-data-warning")
handler = _RecordingHandler()
test_logger.addHandler(handler)
hostile = _HostileValue()
log_model_and_tool_data_warning(
test_logger,
"Fixed operational warning",
diagnostic_message="Warning for %s",
diagnostic_args=lambda: (hostile,),
)
assert len(handler.records) == 1
record = handler.records[0]
assert record.msg == "Fixed operational warning"
assert record.args == ()
assert record.exc_info is None
assert record.exc_text is None
assert hostile not in record.__dict__.values()
assert logging.Formatter().format(record) == "Fixed operational warning"
def test_shared_data_warning_preserves_diagnostics_when_enabled(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
test_logger = logging.Logger("sensitive-logging-data-warning")
handler = _RecordingHandler()
test_logger.addHandler(handler)
diagnostic_value = "diagnostic-agent"
log_model_and_tool_data_warning(
test_logger,
"Fixed operational warning",
diagnostic_message="Warning for %s",
diagnostic_args=lambda: (diagnostic_value,),
)
assert len(handler.records) == 1
record = handler.records[0]
assert record.msg == "Warning for %s"
assert record.args == (diagnostic_value,)
assert logging.Formatter().format(record) == "Warning for diagnostic-agent"
def test_shared_data_warning_falls_back_when_diagnostic_arguments_fail(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
test_logger = logging.Logger("sensitive-logging-data-warning")
handler = _RecordingHandler()
test_logger.addHandler(handler)
def diagnostic_args() -> tuple[object, ...]:
raise RuntimeError("SECRET_DIAGNOSTIC_ARGUMENT_FAILURE")
log_model_and_tool_data_warning(
test_logger,
"Fixed operational warning",
diagnostic_message="Warning for %s",
diagnostic_args=diagnostic_args,
)
assert len(handler.records) == 1
record = handler.records[0]
assert record.msg == "Fixed operational warning"
assert record.args == ()
assert record.exc_info is None
assert "SECRET_DIAGNOSTIC_ARGUMENT_FAILURE" not in logging.Formatter().format(record)
def test_shared_data_warning_preserves_direct_caller_location(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
test_logger = logging.Logger("sensitive-logging-data-warning-location")
handler = _RecordingHandler()
test_logger.addHandler(handler)
_emit_data_warning_for_location(test_logger)
record = handler.records[0]
assert Path(record.pathname).resolve() == Path(__file__).resolve()
assert record.funcName == "_emit_data_warning_for_location"
@pytest.mark.parametrize(
("model_redacted", "tool_redacted"),
[
(True, False),
(False, True),
(True, True),
(False, False),
],
)
@pytest.mark.parametrize(
("scenario", "secrets", "redacted_message"),
[
(
"missing_agent",
("SECRET_AGENT_NAME",),
"Agent not found, skipping item",
),
(
"missing_agent_field",
("SECRET_ITEM_TYPE",),
"Item missing agent field, skipping",
),
(
"missing_handoff_agents",
("SECRET_SOURCE_AGENT", "SECRET_TARGET_AGENT"),
"Skipping handoff output item: could not resolve agents",
),
],
)
def test_run_state_deserialization_warnings_follow_both_data_policies(
monkeypatch,
model_redacted: bool,
tool_redacted: bool,
scenario: str,
secrets: tuple[str, ...],
redacted_message: str,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", model_redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_redacted)
known_agent = Agent(name="KnownAgent")
if scenario == "missing_agent":
item_data = {
"type": "message_output_item",
"agent": secrets[0],
"raw_item": {},
}
elif scenario == "missing_agent_field":
item_data = {
"type": secrets[0],
"raw_item": {},
}
else:
item_data = {
"type": "handoff_output_item",
"agent": "KnownAgent",
"source_agent": secrets[0],
"target_agent": secrets[1],
"raw_item": {},
}
test_logger = logging.Logger("sensitive-logging-run-state")
handler = _RecordingHandler()
test_logger.addHandler(handler)
with patch("agents.run_state.logger", test_logger):
result = _deserialize_items([item_data], {"KnownAgent": known_agent})
assert result == []
assert len(handler.records) == 1
record = handler.records[0]
rendered = logging.Formatter().format(record)
redacted = model_redacted or tool_redacted
if redacted:
assert record.msg == redacted_message
assert record.args == ()
assert record.exc_info is None
assert record.exc_text is None
assert rendered == redacted_message
for secret in secrets:
assert secret not in rendered
assert secret not in record.__dict__.values()
else:
for secret in secrets:
assert secret in rendered
@pytest.mark.parametrize(
"operation",
[
"on_trace_start",
"on_trace_end",
"on_span_start",
"on_span_end",
"force_flush",
"shutdown",
],
)
@pytest.mark.parametrize(
("model_redacted", "tool_redacted"),
[(True, False), (False, True), (False, False)],
)
def test_trace_processor_failure_identity_follows_both_data_policies(
monkeypatch,
operation: str,
model_redacted: bool,
tool_redacted: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", model_redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_redacted)
test_logger = logging.Logger("sensitive-logging-trace-processor", level=logging.DEBUG)
test_logger.propagate = False
handler = _RecordingHandler()
test_logger.addHandler(handler)
failing = _FailingTracingProcessor()
multi = SynchronousMultiTracingProcessor()
multi.add_tracing_processor(failing)
with patch("agents.tracing.provider.logger", test_logger):
if operation.startswith(("on_trace", "on_span")):
getattr(multi, operation)(object())
else:
getattr(multi, operation)()
record = next(record for record in handler.records if record.levelno == logging.ERROR)
redacted = model_redacted or tool_redacted
if redacted:
assert "openai_agents_diagnostic_context" not in record.__dict__
assert failing not in record.__dict__.values()
assert record.exc_info is None
assert _SECRET not in logging.Formatter().format(record)
assert failing.str_calls == 0
else:
processor_identity = record.__dict__["openai_agents_diagnostic_context"]["trace_processor"]
assert isinstance(processor_identity, str)
assert type(failing).__module__ in processor_identity
assert type(failing).__qualname__ in processor_identity
assert f"{id(failing):x}" in processor_identity
prepared = QueueHandler(SimpleQueue()).prepare(record)
pickle.dumps(prepared)
assert record.exc_info is not None
assert record.exc_info[1] is not None
assert _SECRET in logging.Formatter().format(record)
@pytest.mark.parametrize(
"helper",
[log_shared_tool_action_error, log_tool_action_warning],
)
def test_shared_error_helpers_preserve_direct_caller_location(monkeypatch, helper) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
test_logger = logging.Logger("sensitive-logging-location")
handler = _RecordingHandler()
test_logger.addHandler(handler)
_emit_shared_error_for_location(test_logger, helper)
record = handler.records[0]
assert Path(record.pathname).resolve() == Path(__file__).resolve()
assert record.funcName == "_emit_shared_error_for_location"
def test_tool_execution_error_helper_preserves_external_caller_location(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
test_logger = logging.Logger("sensitive-logging-wrapped-location")
handler = _RecordingHandler()
test_logger.addHandler(handler)
with patch("agents.run_internal.tool_execution.logger", test_logger):
_emit_tool_execution_error_for_location()
record = handler.records[0]
assert Path(record.pathname).resolve() == Path(__file__).resolve()
assert record.funcName == "_emit_tool_execution_error_for_location"
def test_shared_error_helper_drops_exception_chains_and_notes(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
test_logger = logging.Logger("sensitive-logging-chain")
handler = _RecordingHandler()
test_logger.addHandler(handler)
cause = ValueError(f"{_SECRET} cause")
error = RuntimeError(f"{_SECRET} outer")
error.__cause__ = cause
if hasattr(error, "add_note"):
error.add_note(f"{_SECRET} note")
else:
error.__notes__ = [f"{_SECRET} note"]
log_model_action_error(test_logger, "Model failed", error)
record = handler.records[0]
assert record.exc_info is None
assert record.exc_text is None
assert error not in record.__dict__.values()
assert _SECRET not in logging.Formatter().format(record)
def test_log_tool_action_error_logs_full_when_tool_data_enabled(monkeypatch) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
with patch("agents.run_internal.tool_execution.logger") as mock_logger:
log_tool_action_error("Shell executor failed", ValueError("rm -rf /secret/path"))
mock_logger.error.assert_called_once()
logged = str(mock_logger.error.call_args)
assert "/secret/path" in logged
exc_info = mock_logger.error.call_args.kwargs.get("exc_info")
assert isinstance(exc_info, tuple)
assert exc_info[0] is ValueError
assert isinstance(exc_info[1], ValueError)
assert exc_info[2] is None
@pytest.mark.asyncio
async def test_approval_rejection_formatter_error_redacts_exception(monkeypatch, caplog) -> None:
caplog.set_level(logging.DEBUG)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
def boom(_args):
raise ValueError("formatter blew up SECRET_FMT_123")
tool_name = "SECRET_FORMATTER_TOOL_NAME"
result = await resolve_approval_rejection_message(
context_wrapper=RunContextWrapper(context=None),
run_config=RunConfig(tool_error_formatter=boom),
tool_type="function",
tool_name=tool_name,
call_id="call_1",
)
assert isinstance(result, str) and result
record = next(
record for record in caplog.records if "Tool error formatter failed" in record.getMessage()
)
assert record.msg == "%s"
assert record.args == ("Tool error formatter failed",)
assert record.exc_info is None
assert "openai_agents_diagnostic_context" not in record.__dict__
assert tool_name not in caplog.text
assert "SECRET_FMT_123" not in caplog.text
@pytest.mark.asyncio
async def test_approval_rejection_formatter_error_logs_full_when_enabled(
monkeypatch, caplog
) -> None:
caplog.set_level(logging.DEBUG)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
def boom(_args):
raise ValueError("formatter blew up SECRET_FMT_123")
tool_name = "diagnostic_tool"
await resolve_approval_rejection_message(
context_wrapper=RunContextWrapper(context=None),
run_config=RunConfig(tool_error_formatter=boom),
tool_type="function",
tool_name=tool_name,
call_id="call_1",
)
record = next(
record for record in caplog.records if "Tool error formatter failed" in record.getMessage()
)
assert record.__dict__["openai_agents_diagnostic_context"] == {"tool_name": tool_name}
assert record.exc_info is not None
assert "SECRET_FMT_123" in caplog.text
_TOOL_ARGUMENT_SECRET = "SECRET_TOOL_ARGUMENT_123"
def _requires_integer_argument(value: int) -> str:
return str(value)
@pytest.mark.asyncio
async def test_function_tool_validation_error_redacts_payload_when_tool_data_disabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
tool = function_tool(_requires_integer_argument, failure_error_function=None)
payload = f'{{"value": "{_TOOL_ARGUMENT_SECRET}"}}'
with pytest.raises(ModelBehaviorError) as exc_info:
await tool.on_invoke_tool(
ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=payload),
payload,
)
error = exc_info.value
assert str(error) == f"Invalid JSON input for tool {tool.name}"
assert _TOOL_ARGUMENT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
@pytest.mark.asyncio
async def test_function_tool_validation_error_preserves_diagnostics_when_tool_data_enabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
tool = function_tool(_requires_integer_argument, failure_error_function=None)
payload = f'{{"value": "{_TOOL_ARGUMENT_SECRET}"}}'
with pytest.raises(ModelBehaviorError) as exc_info:
await tool.on_invoke_tool(
ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=payload),
payload,
)
error = exc_info.value
assert _TOOL_ARGUMENT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
class _AgentToolParameters(BaseModel):
value: int
@pytest.mark.asyncio
async def test_agent_tool_validation_error_redacts_payload_when_tool_data_disabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
tool = Agent(name="worker").as_tool(
tool_name="worker_tool",
tool_description="Runs the worker agent.",
parameters=_AgentToolParameters,
failure_error_function=None,
)
payload = f'{{"value": "{_TOOL_ARGUMENT_SECRET}"}}'
with pytest.raises(ModelBehaviorError) as exc_info:
await tool.on_invoke_tool(
ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=payload),
payload,
)
error = exc_info.value
assert str(error) == f"Invalid JSON input for tool {tool.name}"
assert _TOOL_ARGUMENT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
@pytest.mark.asyncio
async def test_agent_tool_validation_error_preserves_diagnostics_when_tool_data_enabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
tool = Agent(name="worker").as_tool(
tool_name="worker_tool",
tool_description="Runs the worker agent.",
parameters=_AgentToolParameters,
failure_error_function=None,
)
payload = f'{{"value": "{_TOOL_ARGUMENT_SECRET}"}}'
with pytest.raises(ModelBehaviorError) as exc_info:
await tool.on_invoke_tool(
ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments=payload),
payload,
)
error = exc_info.value
assert _TOOL_ARGUMENT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
_TOOL_OUTPUT_SECRET = "SECRET_TOOL_OUTPUT_123"
class _IntegerOutput(BaseModel):
value: int
def _returns_wrong_typed_output(ignored: str = "") -> Any:
# The declared output type expects an ``int`` for ``value``; returning the secret string
# instead triggers output validation, whose ValidationError repr embeds the raw output value.
return {"value": _TOOL_OUTPUT_SECRET}
@pytest.mark.asyncio
async def test_function_tool_output_validation_error_redacts_payload_when_tool_data_disabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
tool = function_tool(
_returns_wrong_typed_output,
name_override="output_tool",
output_type=_IntegerOutput,
failure_error_function=None,
)
with pytest.raises(UserError) as exc_info:
await tool.on_invoke_tool(
ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments="{}"),
"{}",
)
error = exc_info.value
assert _TOOL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(error, _TOOL_OUTPUT_SECRET)
@pytest.mark.asyncio
async def test_function_tool_output_validation_error_preserves_diagnostics_when_enabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
tool = function_tool(
_returns_wrong_typed_output,
name_override="output_tool",
output_type=_IntegerOutput,
failure_error_function=None,
)
with pytest.raises(UserError) as exc_info:
await tool.on_invoke_tool(
ToolContext(None, tool_name=tool.name, tool_call_id="1", tool_arguments="{}"),
"{}",
)
error = exc_info.value
assert _TOOL_OUTPUT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
_MODEL_OUTPUT_SECRET = "SECRET_MODEL_OUTPUT_123"
_SENSITIVE_SCHEMA_SECRET = "SENSITIVE_HANDOFF_SCHEMA_SECRET_4207"
_SENSITIVE_OUTPUT_SCHEMA_SECRET = "SENSITIVE_OUTPUT_SCHEMA_SECRET_4207"
_SensitiveHandoffInput = Literal["SENSITIVE_HANDOFF_SCHEMA_SECRET_4207"]
_SensitiveOutput = Literal["SENSITIVE_OUTPUT_SCHEMA_SECRET_4207"]
class _RequiredOutput(BaseModel):
answer: str
count: int
class _PermissiveFallbackOutput(BaseModel):
payload: SkipValidation[str]
count: int
def _assert_secret_absent_from_agents_traceback(
error: BaseException,
secret: str,
*,
require_agents_frames: bool = True,
) -> None:
traceback_exception = traceback.TracebackException.from_exception(error, capture_locals=True)
agents_source = (Path(__file__).parents[1] / "src" / "agents").resolve()
agents_frames = [
frame
for frame in traceback_exception.stack
if Path(frame.filename).resolve().is_relative_to(agents_source)
]
if require_agents_frames:
assert agents_frames
for frame in agents_frames:
assert secret not in "".join((frame.locals or {}).values())
def _agents_traceback_frame_locals(error: BaseException) -> list[dict[str, Any]]:
agents_source = (Path(__file__).parents[1] / "src" / "agents").resolve()
frame_locals: list[dict[str, Any]] = []
traceback_object = error.__traceback__
while traceback_object is not None:
if (
Path(traceback_object.tb_frame.f_code.co_filename)
.resolve()
.is_relative_to(agents_source)
):
frame_locals.append(traceback_object.tb_frame.f_locals)
traceback_object = traceback_object.tb_next
return frame_locals
def _assert_handoff_closure_absent_from_traceback(
error: BaseException,
*,
callback: Any,
schema_secret: str,
) -> None:
for frame_locals in _agents_traceback_frame_locals(error):
for value in frame_locals.values():
closure = getattr(value, "__closure__", None)
if closure is None:
continue
closure_values = [cell.cell_contents for cell in closure]
assert all(item is not callback for item in closure_values)
assert schema_secret not in repr(closure_values)
@pytest.mark.parametrize(
("model_redacted", "tool_redacted", "expected_redacted"),
[
(True, False, True),
(False, True, False),
],
)
def test_output_schema_validation_error_follows_model_data_policy(
monkeypatch: pytest.MonkeyPatch,
model_redacted: bool,
tool_redacted: bool,
expected_redacted: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", model_redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_redacted)
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
with pytest.raises(ModelBehaviorError) as exc_info:
AgentOutputSchema(_RequiredOutput).validate_json(payload)
error = exc_info.value
if expected_redacted:
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
require_agents_frames=False,
)
else:
assert _MODEL_OUTPUT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
assert any(
frame_locals.get("json_str") == payload
for frame_locals in _agents_traceback_frame_locals(error)
)
def test_output_schema_redaction_survives_trace_attachment_failure(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
with patch(
"agents.util._json.attach_error_to_current_span",
side_effect=RuntimeError("trace attachment failed"),
):
with pytest.raises(ModelBehaviorError) as exc_info:
AgentOutputSchema(_RequiredOutput).validate_json(payload)
error = exc_info.value
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
require_agents_frames=False,
)
def test_output_schema_redaction_omits_sensitive_schema_metadata(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
output_type = cast(type[Any], _SensitiveOutput)
with pytest.raises(ModelBehaviorError) as exc_info:
AgentOutputSchema(output_type).validate_json('"invalid"')
error = exc_info.value
assert _SENSITIVE_OUTPUT_SCHEMA_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_SENSITIVE_OUTPUT_SCHEMA_SECRET,
require_agents_frames=False,
)
@pytest.mark.parametrize(
("model_redacted", "tool_redacted", "expected_redacted"),
[
(True, False, True),
(False, True, True),
(True, True, True),
(False, False, False),
],
)
@pytest.mark.asyncio
async def test_handoff_input_validation_error_follows_mixed_data_policy(
monkeypatch: pytest.MonkeyPatch,
model_redacted: bool,
tool_redacted: bool,
expected_redacted: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", model_redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_redacted)
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
target = Agent(name="target")
handoff_calls = 0
async def on_handoff(_ctx: RunContextWrapper[Any], _input: _RequiredOutput) -> None:
nonlocal handoff_calls
handoff_calls += 1
handoff_obj = handoff(target, input_type=_RequiredOutput, on_handoff=on_handoff)
with pytest.raises(ModelBehaviorError) as exc_info:
await handoff_obj.on_invoke_handoff(RunContextWrapper(None), payload)
assert handoff_calls == 0
error = exc_info.value
if expected_redacted:
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
require_agents_frames=False,
)
else:
assert _MODEL_OUTPUT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
assert any(
frame_locals.get("input_json") == payload
for frame_locals in _agents_traceback_frame_locals(error)
)
@pytest.mark.asyncio
async def test_handoff_redaction_omits_sensitive_schema_metadata(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
target = Agent(name="target")
handoff_calls = 0
async def on_handoff(_ctx: RunContextWrapper[Any], _input: Any) -> None:
nonlocal handoff_calls
handoff_calls += 1
handoff_obj = handoff(
target,
input_type=cast(type[Any], _SensitiveHandoffInput),
on_handoff=on_handoff,
)
with pytest.raises(ModelBehaviorError) as exc_info:
await handoff_obj.on_invoke_handoff(RunContextWrapper(None), '"invalid"')
error = exc_info.value
assert handoff_calls == 0
assert _SENSITIVE_SCHEMA_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_SENSITIVE_SCHEMA_SECRET,
require_agents_frames=False,
)
_assert_handoff_closure_absent_from_traceback(
error,
callback=on_handoff,
schema_secret=_SENSITIVE_SCHEMA_SECRET,
)
@pytest.mark.asyncio
async def test_realtime_handoff_redaction_omits_sensitive_schema_metadata(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", True)
target = RealtimeAgent(name="target")
handoff_calls = 0
async def on_handoff(_ctx: RunContextWrapper[Any], _input: Any) -> None:
nonlocal handoff_calls
handoff_calls += 1
handoff_obj = realtime_handoff(
target,
input_type=cast(type[Any], _SensitiveHandoffInput),
on_handoff=on_handoff,
)
with pytest.raises(ModelBehaviorError) as exc_info:
await handoff_obj.on_invoke_handoff(RunContextWrapper(None), '"invalid"')
error = exc_info.value
assert handoff_calls == 0
assert _SENSITIVE_SCHEMA_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_SENSITIVE_SCHEMA_SECRET,
require_agents_frames=False,
)
_assert_handoff_closure_absent_from_traceback(
error,
callback=on_handoff,
schema_secret=_SENSITIVE_SCHEMA_SECRET,
)
@pytest.mark.parametrize(
("model_redacted", "tool_redacted", "expected_redacted"),
[
(True, False, True),
(False, True, True),
(True, True, True),
(False, False, False),
],
)
@pytest.mark.asyncio
async def test_realtime_handoff_input_validation_error_follows_mixed_data_policy(
monkeypatch: pytest.MonkeyPatch,
model_redacted: bool,
tool_redacted: bool,
expected_redacted: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", model_redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", tool_redacted)
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
target = RealtimeAgent(name="target")
handoff_calls = 0
async def on_handoff(_ctx: RunContextWrapper[Any], _input: _RequiredOutput) -> None:
nonlocal handoff_calls
handoff_calls += 1
handoff_obj = realtime_handoff(target, input_type=_RequiredOutput, on_handoff=on_handoff)
with pytest.raises(ModelBehaviorError) as exc_info:
await handoff_obj.on_invoke_handoff(RunContextWrapper(None), payload)
assert handoff_calls == 0
error = exc_info.value
if expected_redacted:
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
require_agents_frames=False,
)
else:
assert _MODEL_OUTPUT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
assert any(
frame_locals.get("input_json") == payload
for frame_locals in _agents_traceback_frame_locals(error)
)
@pytest.mark.asyncio
async def test_run_surfaces_redacted_output_validation_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
model = ScriptedModel()
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
model.enqueue([get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')])
session = SimpleListSession(
session_id="redacted-run",
history=[{"role": "user", "content": _MODEL_OUTPUT_SECRET}],
)
with pytest.raises(ModelBehaviorError) as exc_info:
await Runner.run(agent, _MODEL_OUTPUT_SECRET, session=session)
error = exc_info.value
frame_locals = _agents_traceback_frame_locals(error)
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
)
assert all(session is not value for frame in frame_locals for value in frame.values())
def test_run_sync_surfaces_redacted_output_validation_error_without_runner_data(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
model = ScriptedModel()
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
model.enqueue([get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')])
session = SimpleListSession(
session_id="redacted-run-sync",
history=[{"role": "user", "content": _MODEL_OUTPUT_SECRET}],
)
with pytest.raises(ModelBehaviorError) as exc_info:
Runner.run_sync(agent, _MODEL_OUTPUT_SECRET, session=session)
error = exc_info.value
frame_locals = _agents_traceback_frame_locals(error)
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
)
assert all(session is not value for frame in frame_locals for value in frame.values())
def test_run_sync_preserves_redacted_hybrid_cancellation_catchability(
monkeypatch: pytest.MonkeyPatch,
) -> None:
hybrid_error = _HybridCancelledError("RUN_SYNC_HYBRID_SECRET")
_mark_error_data_redacted(hybrid_error)
async def raise_hybrid_error(*_args: Any, **_kwargs: Any) -> Any:
raise hybrid_error
monkeypatch.setattr(AgentRunner, "run", raise_hybrid_error)
with pytest.raises(Exception) as exc_info:
AgentRunner().run_sync(Agent(name="A"), "RUN_SYNC_INPUT_SECRET")
error = exc_info.value
assert isinstance(error, asyncio.CancelledError)
assert isinstance(error, Exception)
assert str(error) == ""
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(error, "RUN_SYNC_HYBRID_SECRET")
_assert_secret_absent_from_agents_traceback(error, "RUN_SYNC_INPUT_SECRET")
@pytest.mark.asyncio
async def test_run_preserves_diagnostic_wrapper_traceback_locals(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
diagnostic_input = "DIAGNOSTIC_RUNNER_INPUT_SECRET"
model = ScriptedModel(steps=[[get_text_message('{"answer": "missing count"}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
session = SimpleListSession(session_id="diagnostic-runner")
with pytest.raises(ModelBehaviorError) as exc_info:
await Runner.run(agent, diagnostic_input, session=session)
error = exc_info.value
frame_locals = _agents_traceback_frame_locals(error)
assert any(frame.get("input") == diagnostic_input for frame in frame_locals)
assert any(frame.get("session") is session for frame in frame_locals)
@pytest.mark.asyncio
async def test_streaming_redaction_boundary_defers_inner_coroutine_until_task_starts(
monkeypatch: pytest.MonkeyPatch,
) -> None:
start_streaming_called = False
async def inner_streaming() -> None:
return None
def start_streaming_factory(**_kwargs: Any) -> Any:
nonlocal start_streaming_called
start_streaming_called = True
return inner_streaming()
monkeypatch.setattr("agents.run.start_streaming", start_streaming_factory)
result = AgentRunner().run_streamed(Agent(name="A", model=ScriptedModel()), "go")
assert result.run_loop_task is not None
result.run_loop_task.cancel()
with pytest.raises(asyncio.CancelledError):
await result.run_loop_task
assert not start_streaming_called
def test_run_sync_preserves_diagnostic_wrapper_traceback_locals(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
diagnostic_input = "DIAGNOSTIC_RUNNER_SYNC_INPUT_SECRET"
model = ScriptedModel(steps=[[get_text_message('{"answer": "missing count"}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
session = SimpleListSession(session_id="diagnostic-runner-sync")
with pytest.raises(ModelBehaviorError) as exc_info:
Runner.run_sync(agent, diagnostic_input, session=session)
error = exc_info.value
frame_locals = _agents_traceback_frame_locals(error)
assert any(frame.get("input") == diagnostic_input for frame in frame_locals)
assert any(frame.get("session") is session for frame in frame_locals)
@pytest.mark.asyncio
async def test_streamed_run_surfaces_redacted_output_validation_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
model = ScriptedModel()
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
model.enqueue([get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')])
result = Runner.run_streamed(agent, "go")
with pytest.raises(ModelBehaviorError) as exc_info:
async for _ in result.stream_events():
pass
error = exc_info.value
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
@pytest.mark.parametrize("redacted", [False, True])
@pytest.mark.asyncio
async def test_streamed_run_loop_exception_follows_model_data_policy(
monkeypatch: pytest.MonkeyPatch,
redacted: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", redacted)
model = ScriptedModel(steps=[[get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
result = Runner.run_streamed(agent, "go")
assert result.run_loop_task is not None
while not result.run_loop_task.done():
await asyncio.sleep(0)
error = result.run_loop_exception
assert isinstance(error, ModelBehaviorError)
if redacted:
assert error.run_data is None
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
require_agents_frames=False,
)
else:
assert _MODEL_OUTPUT_SECRET in str(error)
assert isinstance(error.__cause__, ValidationError)
assert any(
_MODEL_OUTPUT_SECRET in repr(frame_locals)
for frame_locals in _agents_traceback_frame_locals(error)
)
@pytest.mark.asyncio
async def test_streamed_output_guardrail_omits_run_data_from_redacted_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
model = ScriptedModel(steps=[[get_text_message(_MODEL_OUTPUT_SECRET)]])
agent = Agent(
name="A",
model=model,
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
result = Runner.run_streamed(agent, "go")
with pytest.raises(ModelBehaviorError) as exc_info:
async for _ in result.stream_events():
pass
error = exc_info.value
assert error.run_data is None
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
@pytest.mark.parametrize("redacted", [False, True])
@pytest.mark.parametrize("persistence_failure", ["error", "cancelled"])
@pytest.mark.asyncio
async def test_streamed_session_error_after_output_guardrail_respects_redaction(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
redacted: bool,
persistence_failure: Literal["error", "cancelled"],
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
guardrail_failed = False
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
class FailingFinalTurnSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
cause = RuntimeError(f"session save cause: {_MODEL_OUTPUT_SECRET}")
if persistence_failure == "cancelled":
raise asyncio.CancelledError(
f"session save cancelled: {_MODEL_OUTPUT_SECRET}"
) from cause
raise LookupError(f"session save failed: {_MODEL_OUTPUT_SECRET}") from cause
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
caplog.set_level(logging.ERROR, logger="openai.agents")
model = ScriptedModel(steps=[[get_text_message(_MODEL_OUTPUT_SECRET)]])
agent = Agent(
name="A",
model=model,
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
result = Runner.run_streamed(agent, "go", session=FailingFinalTurnSession())
run_loop_callback_errors: list[BaseException] = []
run_loop_done = asyncio.Event()
if redacted and persistence_failure == "cancelled":
assert result.run_loop_task is not None
def inspect_run_loop_task(task: asyncio.Task[Any]) -> None:
try:
task.result()
except BaseException as error:
run_loop_callback_errors.append(error)
finally:
run_loop_done.set()
result.run_loop_task.add_done_callback(inspect_run_loop_task)
expected_error_type = (
asyncio.CancelledError
if persistence_failure == "cancelled"
else UserError
if redacted
else LookupError
)
expected_message = (
"Error details are redacted."
if redacted
else "session save cancelled"
if persistence_failure == "cancelled"
else "session save failed"
)
with pytest.raises(expected_error_type, match=expected_message) as exc_info:
async for _ in result.stream_events():
pass
error = exc_info.value
guardrail_error = error.__context__
if redacted:
assert guardrail_error is None
assert error.__cause__ is None
assert _MODEL_OUTPUT_SECRET not in str(error)
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
for record in caplog.records:
assert _MODEL_OUTPUT_SECRET not in repr(record.__dict__)
assert _MODEL_OUTPUT_SECRET not in logging.Formatter().format(record)
assert record.exc_info is None
else:
assert isinstance(guardrail_error, ModelBehaviorError)
assert error.__cause__ is not None
assert _MODEL_OUTPUT_SECRET in str(error.__cause__)
assert _MODEL_OUTPUT_SECRET in str(guardrail_error)
assert any(
_MODEL_OUTPUT_SECRET in repr(frame) for frame in _agents_traceback_frame_locals(error)
)
if persistence_failure == "cancelled":
assert result._stored_exception is error
assert result.run_loop_exception is None
if redacted:
await asyncio.wait_for(run_loop_done.wait(), timeout=1)
assert run_loop_callback_errors == []
else:
assert result.run_loop_exception is error
if redacted:
assert error.__traceback__ is None
@pytest.mark.asyncio
async def test_streamed_session_hostile_error_after_redacted_output_guardrail_is_replaced(
monkeypatch: pytest.MonkeyPatch,
) -> None:
persistence_secret = "HOSTILE_SESSION_FAILURE_SECRET"
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
guardrail_failed = False
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
class FailingFinalTurnSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
raise _HostileAttributeWriteException(persistence_secret)
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
agent = Agent(
name="A",
model=ScriptedModel(steps=[[get_text_message(_MODEL_OUTPUT_SECRET)]]),
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
result = Runner.run_streamed(agent, "go", session=FailingFinalTurnSession())
with pytest.raises(UserError, match="Error details are redacted.") as exc_info:
async for _ in result.stream_events():
pass
error = exc_info.value
assert error.__cause__ is None
assert error.__context__ is None
assert persistence_secret not in str(error)
_assert_secret_absent_from_agents_traceback(error, persistence_secret)
_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,
) -> BaseException:
if kind == "exception":
return LookupError(f"session save failed: {secret}")
if kind == "cancelled":
return asyncio.CancelledError(f"session save cancelled: {secret}")
if kind == "direct_base":
return _DirectBaseException(f"session save aborted: {secret}")
if kind == "exception_group":
return BaseExceptionGroup(
f"session save exception group: {secret}",
[RuntimeError(f"session save child: {secret}")],
)
return BaseExceptionGroup(
f"session save group: {secret}",
[
RuntimeError(f"session save child: {secret}"),
asyncio.CancelledError(f"session save cancelled child: {secret}"),
],
)
def _exception_graph(error: BaseException) -> list[BaseException]:
graph: list[BaseException] = []
pending = [error]
seen: set[int] = set()
while pending:
current = pending.pop()
if id(current) in seen:
continue
seen.add(id(current))
graph.append(current)
if isinstance(current, BaseExceptionGroup):
pending.extend(current.exceptions)
if current.__cause__ is not None:
pending.append(current.__cause__)
if current.__context__ is not None:
pending.append(current.__context__)
return graph
def _assert_secret_absent_from_value_graph(value: Any, secret: str) -> None:
pending = [value]
seen: set[int] = set()
while pending:
current = pending.pop()
if id(current) in seen:
continue
seen.add(id(current))
if isinstance(current, str):
assert secret not in current
elif isinstance(current, bytes):
assert secret.encode() not in current
elif isinstance(current, BaseExceptionGroup):
assert secret not in current.message
pending.extend(current.exceptions)
pending.extend(current.args)
elif isinstance(current, BaseException):
assert secret not in str(current)
assert secret not in repr(current)
pending.extend(current.args)
elif type(current) is dict:
pending.extend(current.keys())
pending.extend(current.values())
elif type(current) in {list, tuple, set, frozenset}:
pending.extend(current)
else:
assert secret not in repr(current)
@pytest.mark.parametrize("redacted", [False, True])
@pytest.mark.parametrize(
"failure_kind",
["exception", "cancelled", "direct_base", "exception_group", "base_group"],
)
@pytest.mark.asyncio
async def test_sandbox_cleanup_wrapper_preserves_redaction_boundary(
redacted: bool,
failure_kind: Literal["exception", "cancelled", "direct_base", "exception_group", "base_group"],
) -> None:
secret = f"SANDBOX_WRAPPER_SECRET_{failure_kind}"
if failure_kind == "exception":
error: BaseException = RuntimeError()
elif failure_kind == "cancelled":
error = asyncio.CancelledError()
elif failure_kind == "direct_base":
error = _DirectBaseException()
elif failure_kind == "exception_group":
error = BaseExceptionGroup("Error details are redacted.", [RuntimeError()])
else:
error = BaseExceptionGroup(
"Error details are redacted.",
[asyncio.CancelledError()],
)
async def original_task() -> None:
payload = secret
assert payload
try:
raise error
except BaseException as caught:
if redacted:
_mark_error_data_redacted(caught)
_detach_data_redacted_error_traceback(caught)
payload = None
raise
async def cleanup() -> None:
return None
result = RunResultStreaming(
input=secret,
new_items=[],
raw_responses=[],
final_output=None,
input_guardrail_results=[],
output_guardrail_results=[],
tool_input_guardrail_results=[],
tool_output_guardrail_results=[],
context_wrapper=RunContextWrapper(context=None),
current_agent=Agent(name="test"),
current_turn=0,
max_turns=1,
_current_agent_output_schema=None,
trace=None,
)
result._sandbox_cleanup = cleanup
result.run_loop_task = asyncio.create_task(original_task())
result.ensure_sandbox_cleanup_on_completion()
assert result.run_loop_task is not None
callback_frame_locals: list[dict[str, Any]] = []
task_done = asyncio.Event()
def inspect_public_task(task: asyncio.Task[Any]) -> None:
try:
task.result()
except BaseException as caught:
callback_frame_locals.extend(_agents_traceback_frame_locals(caught))
finally:
task_done.set()
result.run_loop_task.add_done_callback(inspect_public_task)
await asyncio.wait_for(task_done.wait(), timeout=1)
if redacted:
for frame_locals in callback_frame_locals:
_assert_secret_absent_from_value_graph(frame_locals, secret)
else:
if failure_kind == "cancelled" and sys.version_info < (3, 11):
assert callback_frame_locals == []
else:
assert callback_frame_locals
assert any(secret in repr(frame) for frame in callback_frame_locals)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.parametrize("failure_kind", ["exception", "cancelled", "direct_base", "group"])
@pytest.mark.asyncio
async def test_max_turns_recovery_session_failure_preserves_complete_redaction_boundary(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
streamed: bool,
failure_kind: Literal["exception", "cancelled", "direct_base", "group"],
) -> None:
persistence_secret = "MAX_TURNS_SESSION_FAILURE_SECRET"
fallback_secret = "MAX_TURNS_HANDLER_OUTPUT_SECRET"
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
guardrail_failed = False
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
class FailingMaxTurnsSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
raise _persistence_failure(failure_kind, persistence_secret)
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
caplog.set_level(logging.ERROR, logger="openai.agents")
agent = Agent(
name="A",
model=ScriptedModel(),
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
session = FailingMaxTurnsSession()
captured_error: BaseException | None = None
try:
if streamed:
result = Runner.run_streamed(
agent,
"go",
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: fallback_secret},
)
async for _ in result.stream_events():
pass
else:
await Runner.run(
agent,
"go",
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: fallback_secret},
)
except BaseException as error:
captured_error = error
else: # pragma: no cover
raise AssertionError("the session failure must propagate")
assert captured_error is not None
error = captured_error
if failure_kind == "exception":
assert isinstance(error, UserError)
elif failure_kind == "cancelled":
assert isinstance(error, asyncio.CancelledError)
elif failure_kind == "direct_base":
assert type(error) is BaseException
else:
assert isinstance(error, BaseExceptionGroup)
assert not isinstance(error, Exception)
assert {type(child) for child in error.exceptions} == {
UserError,
asyncio.CancelledError,
}
error_graph = _exception_graph(error)
assert error_graph
for current in error_graph:
assert current.__cause__ is None
assert current.__context__ is None
assert persistence_secret not in str(current)
assert persistence_secret not in repr(current)
assert fallback_secret not in str(current)
assert fallback_secret not in repr(current)
assert _MODEL_OUTPUT_SECRET not in str(current)
assert _MODEL_OUTPUT_SECRET not in repr(current)
_assert_secret_absent_from_agents_traceback(
current,
persistence_secret,
require_agents_frames=False,
)
_assert_secret_absent_from_agents_traceback(
current,
fallback_secret,
require_agents_frames=False,
)
_assert_secret_absent_from_agents_traceback(
current,
_MODEL_OUTPUT_SECRET,
require_agents_frames=False,
)
for record in caplog.records:
rendered_record = logging.Formatter().format(record)
record_state = repr(record.__dict__)
for secret in (persistence_secret, fallback_secret, _MODEL_OUTPUT_SECRET):
assert secret not in rendered_record
assert secret not in record_state
assert record.exc_info is None
def _direct_agent_runner_redaction_case(
monkeypatch: pytest.MonkeyPatch,
failure_kind: Literal["cancelled", "direct_base", "exception_group", "group"],
) -> tuple[Agent[Any], SimpleListSession, str, tuple[str, str, str, str]]:
persistence_secret = f"DIRECT_AGENT_RUNNER_PERSISTENCE_SECRET_{failure_kind}"
fallback_secret = f"DIRECT_AGENT_RUNNER_FALLBACK_SECRET_{failure_kind}"
input_secret = f"DIRECT_AGENT_RUNNER_INPUT_SECRET_{failure_kind}"
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
guardrail_failed = False
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
class FailingSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
raise _persistence_failure(failure_kind, persistence_secret)
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
agent = Agent(
name="A",
model=ScriptedModel(),
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
secrets = (persistence_secret, fallback_secret, input_secret, _MODEL_OUTPUT_SECRET)
return agent, FailingSession(), input_secret, secrets
def _assert_direct_agent_runner_redaction_boundary(
error: BaseException,
secrets: tuple[str, str, str, str],
) -> None:
for current in _exception_graph(error):
assert current.__cause__ is None
assert current.__context__ is None
for secret in secrets:
assert secret not in str(current)
assert secret not in repr(current)
for frame_locals in _agents_traceback_frame_locals(current):
_assert_secret_absent_from_value_graph(frame_locals, secret)
@pytest.mark.parametrize(
"failure_kind",
["cancelled", "direct_base", "exception_group", "group"],
)
@pytest.mark.asyncio
async def test_agent_runner_run_detaches_all_marked_recovery_failures(
monkeypatch: pytest.MonkeyPatch,
failure_kind: Literal["cancelled", "direct_base", "exception_group", "group"],
) -> None:
agent, session, input_secret, secrets = _direct_agent_runner_redaction_case(
monkeypatch, failure_kind
)
with pytest.raises(BaseException) as exc_info:
await AgentRunner().run(
agent,
input_secret,
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: secrets[1]},
)
_assert_direct_agent_runner_redaction_boundary(exc_info.value, secrets)
@pytest.mark.parametrize(
"failure_kind",
["cancelled", "direct_base", "exception_group", "group"],
)
def test_agent_runner_run_sync_detaches_all_marked_recovery_failures(
monkeypatch: pytest.MonkeyPatch,
failure_kind: Literal["cancelled", "direct_base", "exception_group", "group"],
) -> None:
agent, session, input_secret, secrets = _direct_agent_runner_redaction_case(
monkeypatch, failure_kind
)
with pytest.raises(BaseException) as exc_info:
AgentRunner().run_sync(
agent,
input_secret,
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: secrets[1]},
)
_assert_direct_agent_runner_redaction_boundary(exc_info.value, secrets)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_max_turns_recovery_deep_exception_group_preserves_redaction_boundary(
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
streamed: bool,
) -> None:
persistence_secret = "DEEP_MAX_TURNS_SESSION_FAILURE_SECRET"
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
guardrail_failed = False
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
class DeepGroupSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
error: BaseException = asyncio.CancelledError(persistence_secret)
for _ in range(1200):
error = BaseExceptionGroup(persistence_secret, [error])
raise error
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
caplog.set_level(logging.ERROR, logger="openai.agents")
agent = Agent(
name="A",
model=ScriptedModel(),
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
session = DeepGroupSession()
captured_error: BaseException | None = None
try:
if streamed:
result = Runner.run_streamed(
agent,
"go",
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: "fallback"},
)
async for _ in result.stream_events():
pass
else:
await Runner.run(
agent,
"go",
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: "fallback"},
)
except BaseException as error:
captured_error = error
else: # pragma: no cover
raise AssertionError("the deep exception group must propagate")
assert isinstance(captured_error, BaseExceptionGroup)
error_graph = _exception_graph(captured_error)
assert len(error_graph) == 1201
for current in error_graph:
assert current.__cause__ is None
assert current.__context__ is None
if isinstance(current, BaseExceptionGroup):
assert current.message == "Error details are redacted."
else:
assert isinstance(current, asyncio.CancelledError)
assert persistence_secret not in str(current)
assert persistence_secret not in repr(current)
for frame_locals in _agents_traceback_frame_locals(current):
_assert_secret_absent_from_value_graph(frame_locals, persistence_secret)
_assert_secret_absent_from_value_graph(frame_locals, _MODEL_OUTPUT_SECRET)
for record in caplog.records:
assert persistence_secret not in logging.Formatter().format(record)
assert persistence_secret not in repr(record.__dict__)
assert _MODEL_OUTPUT_SECRET not in logging.Formatter().format(record)
assert _MODEL_OUTPUT_SECRET not in repr(record.__dict__)
assert record.exc_info is None
@pytest.mark.parametrize("redacted", [False, True])
@pytest.mark.parametrize("failure_kind", ["direct_base", "group"])
@pytest.mark.asyncio
async def test_max_turns_run_loop_exception_follows_redaction_policy_for_base_exceptions(
monkeypatch: pytest.MonkeyPatch,
redacted: bool,
failure_kind: Literal["direct_base", "group"],
) -> None:
persistence_secret = "RUN_LOOP_EXCEPTION_PERSISTENCE_SECRET"
fallback_secret = "RUN_LOOP_EXCEPTION_FALLBACK_SECRET"
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
guardrail_failed = False
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
class FailingMaxTurnsSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
raise _persistence_failure(failure_kind, persistence_secret)
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
result = Runner.run_streamed(
Agent(
name="A",
model=ScriptedModel(),
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
),
"go",
max_turns=0,
session=FailingMaxTurnsSession(),
error_handlers={"max_turns": lambda data: fallback_secret},
)
assert result.run_loop_task is not None
callback_frame_locals: list[dict[str, Any]] = []
run_loop_done = asyncio.Event()
def inspect_run_loop_task(task: asyncio.Task[Any]) -> None:
try:
task.result()
except BaseException as error:
callback_frame_locals.extend(_agents_traceback_frame_locals(error))
finally:
run_loop_done.set()
result.run_loop_task.add_done_callback(inspect_run_loop_task)
await asyncio.wait_for(run_loop_done.wait(), timeout=1)
error = result.run_loop_exception
assert error is not None
frame_locals = _agents_traceback_frame_locals(error)
if redacted:
for traceback_locals in callback_frame_locals + frame_locals:
for secret in (persistence_secret, fallback_secret, _MODEL_OUTPUT_SECRET):
_assert_secret_absent_from_value_graph(traceback_locals, secret)
for current in _exception_graph(error):
assert current.__cause__ is None
assert current.__context__ is None
if isinstance(current, BaseExceptionGroup):
assert current.message == "Error details are redacted."
else:
for secret in (persistence_secret, fallback_secret, _MODEL_OUTPUT_SECRET):
assert secret not in str(current)
assert secret not in repr(current)
else:
assert callback_frame_locals
assert any(fallback_secret in repr(frame) for frame in callback_frame_locals)
assert frame_locals
assert any(fallback_secret in repr(frame) for frame in frame_locals)
assert persistence_secret in str(error)
try:
async for _ in result.stream_events():
pass
except BaseException as streamed_error:
assert streamed_error is error
else: # pragma: no cover
raise AssertionError("the session failure must propagate through the stream")
@pytest.mark.parametrize(
("source", "expected_type"),
[
(asyncio.CancelledError("secret"), asyncio.CancelledError),
(GeneratorExit("secret"), GeneratorExit),
(KeyboardInterrupt("secret"), KeyboardInterrupt),
(SystemExit("secret"), SystemExit),
(_DirectBaseException("secret"), BaseException),
],
)
def test_safe_redacted_persistence_error_preserves_process_control_semantics(
source: BaseException,
expected_type: type[BaseException],
) -> None:
safe_error = _safe_redacted_persistence_error(source)
assert type(safe_error) is expected_type
assert "secret" not in str(safe_error)
assert "secret" not in repr(safe_error)
assert safe_error.__cause__ is None
assert safe_error.__context__ is None
assert safe_error.__traceback__ is None
def test_safe_redacted_persistence_error_snapshots_linked_group_topology() -> None:
linked_group = BaseExceptionGroup(
"linked group secret",
[KeyboardInterrupt("process-control secret")],
)
first_child = RuntimeError("first child secret")
first_child.__context__ = linked_group
source = BaseExceptionGroup("root group secret", [first_child, linked_group])
safe_error = _safe_redacted_persistence_error(source)
assert isinstance(safe_error, BaseExceptionGroup)
assert len(safe_error.exceptions) == 2
assert isinstance(safe_error.exceptions[0], UserError)
safe_linked_group = safe_error.exceptions[1]
assert isinstance(safe_linked_group, BaseExceptionGroup)
assert len(safe_linked_group.exceptions) == 1
assert type(safe_linked_group.exceptions[0]) is KeyboardInterrupt
assert "secret" not in repr(safe_error)
def test_safe_redacted_persistence_error_preserves_provider_group_catch_category() -> None:
class ProviderBaseExceptionGroup(BaseExceptionGroup):
pass
direct_source = ProviderBaseExceptionGroup(
"direct provider secret",
[RuntimeError("direct leaf secret")],
)
nested_source = BaseExceptionGroup(
"root group secret",
[
KeyboardInterrupt("process-control secret"),
ProviderBaseExceptionGroup(
"nested provider secret",
[RuntimeError("nested leaf secret")],
),
],
)
direct = _safe_redacted_persistence_error(direct_source)
nested = _safe_redacted_persistence_error(nested_source)
assert isinstance(direct, BaseExceptionGroup)
assert not isinstance(direct, Exception)
assert len(direct.exceptions) == 1
assert isinstance(direct.exceptions[0], UserError)
assert isinstance(nested, BaseExceptionGroup)
assert not isinstance(nested, Exception)
assert len(nested.exceptions) == 2
nested_provider = nested.exceptions[1]
assert isinstance(nested_provider, BaseExceptionGroup)
assert not isinstance(nested_provider, Exception)
assert len(nested_provider.exceptions) == 1
assert isinstance(nested_provider.exceptions[0], UserError)
assert "secret" not in repr(direct)
assert "secret" not in repr(nested)
def test_safe_redacted_persistence_error_snapshots_linked_system_exit_code() -> None:
system_exit = SystemExit(7)
first_child = RuntimeError("first child secret")
first_child.__context__ = system_exit
source = BaseExceptionGroup("root group secret", [first_child, system_exit])
safe_error = _safe_redacted_persistence_error(source)
assert isinstance(safe_error, BaseExceptionGroup)
assert isinstance(safe_error.exceptions[0], UserError)
safe_system_exit = safe_error.exceptions[1]
assert type(safe_system_exit) is SystemExit
assert safe_system_exit.code == 7
assert safe_system_exit.args == (7,)
assert "secret" not in repr(safe_error)
def test_safe_redacted_persistence_error_avoids_hostile_class_descriptor() -> None:
source = _HostileClassBaseException("persistence secret")
safe_error = _safe_redacted_persistence_error(source)
assert type(safe_error) is BaseException
assert safe_error.__cause__ is None
assert safe_error.__context__ is None
assert "secret" not in str(safe_error)
assert "secret" not in repr(safe_error)
def test_safe_redacted_persistence_error_preserves_hybrid_cancellation() -> None:
direct_source = _HybridCancelledError("direct secret")
grouped_source = BaseExceptionGroup("group secret", [_HybridCancelledError("child secret")])
direct = _safe_redacted_persistence_error(direct_source)
grouped = _safe_redacted_persistence_error(grouped_source)
assert isinstance(direct, asyncio.CancelledError)
assert isinstance(direct, Exception) is isinstance(direct_source, Exception)
assert str(direct) == "Error details are redacted."
assert isinstance(grouped, BaseExceptionGroup)
assert isinstance(grouped, Exception) is isinstance(grouped_source, Exception)
assert len(grouped.exceptions) == 1
assert isinstance(grouped.exceptions[0], asyncio.CancelledError)
assert isinstance(grouped.exceptions[0], Exception)
assert "secret" not in repr(grouped)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_max_turns_recovery_session_failure_preserves_diagnostic_context(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
) -> None:
persistence_secret = "DIAGNOSTIC_MAX_TURNS_SESSION_SECRET"
guardrail_secret = "DIAGNOSTIC_MAX_TURNS_GUARDRAIL_SECRET"
guardrail_failed = False
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
class FailingMaxTurnsSession(SimpleListSession):
async def add_items(self, items: list[Any]) -> None:
if guardrail_failed:
raise LookupError(persistence_secret)
await super().add_items(items)
def output_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_agent_output: Any,
) -> GuardrailFunctionOutput:
nonlocal guardrail_failed
guardrail_failed = True
raise RuntimeError(guardrail_secret)
agent = Agent(
name="A",
model=ScriptedModel(),
output_guardrails=[OutputGuardrail(guardrail_function=output_guardrail)],
)
session = FailingMaxTurnsSession()
if streamed:
result = Runner.run_streamed(
agent,
"go",
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: "fallback"},
)
with pytest.raises(LookupError, match=persistence_secret) as exc_info:
async for _ in result.stream_events():
pass
else:
with pytest.raises(LookupError, match=persistence_secret) as exc_info:
await Runner.run(
agent,
"go",
max_turns=0,
session=session,
error_handlers={"max_turns": lambda data: "fallback"},
)
guardrail_error = exc_info.value.__context__
assert isinstance(guardrail_error, RuntimeError)
assert guardrail_secret in str(guardrail_error)
@pytest.mark.asyncio
async def test_streamed_input_guardrail_omits_run_data_from_redacted_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", False)
payload = f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}'
def input_guardrail(
_context: RunContextWrapper[Any],
_agent: Agent[Any],
_input: str | list[Any],
) -> GuardrailFunctionOutput:
AgentOutputSchema(_RequiredOutput).validate_json(payload)
raise AssertionError("validation should fail") # pragma: no cover
model = ScriptedModel(steps=[[get_text_message("unused")]])
agent = Agent(
name="A",
model=model,
input_guardrails=[InputGuardrail(guardrail_function=input_guardrail)],
)
result = Runner.run_streamed(agent, _MODEL_OUTPUT_SECRET)
with pytest.raises(ModelBehaviorError) as exc_info:
async for _ in result.stream_events():
pass
error = exc_info.value
assert error.run_data is None
assert _MODEL_OUTPUT_SECRET not in str(error)
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_invalid_final_output_handler_receives_detached_redacted_error(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
model = ScriptedModel(steps=[[get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
retained_errors: list[ModelBehaviorError] = []
def recover(data: RunErrorHandlerInput[None]) -> _RequiredOutput:
assert isinstance(data.error, ModelBehaviorError)
retained_errors.append(data.error)
return _RequiredOutput(answer="safe", count=1)
if streamed:
result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": recover},
)
async for _ in result.stream_events():
pass
else:
await Runner.run(
agent,
"go",
error_handlers={"invalid_final_output": recover},
)
assert len(retained_errors) == 1
error = retained_errors[0]
assert error.__traceback__ is None
assert error.__cause__ is None
assert error.__context__ is None
assert _MODEL_OUTPUT_SECRET not in str(error)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_invalid_final_output_handler_invalid_fallback_preserves_redaction(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
) -> None:
fallback_secret = "INVALID_HANDLER_FALLBACK_SECRET"
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
model = ScriptedModel(steps=[[get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
def invalid_fallback(_data: RunErrorHandlerInput[None]) -> dict[str, str]:
return {"answer": fallback_secret}
with warnings.catch_warnings(record=True) as caught_warnings:
warnings.simplefilter("always")
if streamed:
result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": invalid_fallback},
)
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,
"go",
error_handlers={"invalid_final_output": invalid_fallback},
)
error = exc_info.value
assert not caught_warnings
assert str(error) == "Error details are redacted."
assert error.run_data is None
assert error.__cause__ is None
assert error.__context__ is None
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
_assert_secret_absent_from_agents_traceback(error, fallback_secret)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.parametrize("redacted", [True, False])
@pytest.mark.asyncio
async def test_invalid_final_output_handler_fallback_serialization_follows_redaction_policy(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
redacted: bool,
) -> None:
fallback_secret = "PERMISSIVE_HANDLER_FALLBACK_SECRET"
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", redacted)
model = ScriptedModel(
steps=[[get_text_message(f'{{"payload": "{_MODEL_OUTPUT_SECRET}", "count": "invalid"}}')]]
)
agent = Agent(name="A", model=model, output_type=_PermissiveFallbackOutput)
def permissive_fallback(_data: RunErrorHandlerInput[None]) -> _PermissiveFallbackOutput:
return _PermissiveFallbackOutput(
payload=cast(Any, {"secret": fallback_secret}),
count=1,
)
with warnings.catch_warnings(record=True) as caught_warnings:
warnings.simplefilter("always")
if streamed:
streaming_result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": permissive_fallback},
)
async for _ in streaming_result.stream_events():
pass
actual_final_output = streaming_result.final_output
else:
run_result = await Runner.run(
agent,
"go",
error_handlers={"invalid_final_output": permissive_fallback},
)
actual_final_output = run_result.final_output
rendered_warnings = "\n".join(str(warning.message) for warning in caught_warnings)
if redacted:
assert not caught_warnings
else:
assert fallback_secret in rendered_warnings
assert actual_final_output == _PermissiveFallbackOutput(
payload=cast(Any, {"secret": fallback_secret}),
count=1,
)
@pytest.mark.parametrize(
("streamed", "include_in_history", "redacted"),
[
(False, True, True),
(False, False, True),
(True, True, True),
(True, False, True),
(False, True, False),
],
)
@pytest.mark.asyncio
async def test_empty_final_output_handler_fallback_serialization_follows_redaction_policy(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
include_in_history: bool,
redacted: bool,
) -> None:
fallback_secret = "EMPTY_HANDLER_FALLBACK_SECRET"
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", redacted)
model = ScriptedModel(steps=[[]])
agent = Agent(name="A", model=model, output_type=_PermissiveFallbackOutput)
def permissive_fallback(_data: RunErrorHandlerInput[None]) -> RunErrorHandlerResult:
return RunErrorHandlerResult(
final_output=_PermissiveFallbackOutput(
payload=cast(Any, {"secret": fallback_secret}),
count=1,
),
include_in_history=include_in_history,
)
with warnings.catch_warnings(record=True) as caught_warnings:
warnings.simplefilter("always")
if streamed:
streaming_result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": permissive_fallback},
)
async for _ in streaming_result.stream_events():
pass
actual_final_output = streaming_result.final_output
else:
run_result = await Runner.run(
agent,
"go",
error_handlers={"invalid_final_output": permissive_fallback},
)
actual_final_output = run_result.final_output
rendered_warnings = "\n".join(str(warning.message) for warning in caught_warnings)
if redacted:
assert not caught_warnings
else:
assert fallback_secret in rendered_warnings
assert actual_final_output == _PermissiveFallbackOutput(
payload=cast(Any, {"secret": fallback_secret}),
count=1,
)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_invalid_final_output_handler_failure_preserves_redaction(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
model = ScriptedModel(steps=[[get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
def fail(data: RunErrorHandlerInput[None]) -> None:
raise RuntimeError(repr(data.run_data.raw_responses))
if streamed:
result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": fail},
)
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,
"go",
error_handlers={"invalid_final_output": fail},
)
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
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_invalid_final_output_handler_hostile_failure_preserves_redaction(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
) -> None:
handler_secret = "HOSTILE_HANDLER_FAILURE_SECRET"
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", True)
model = ScriptedModel(steps=[[get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
def fail(_data: RunErrorHandlerInput[None]) -> None:
raise _HostileAttributeWriteException(handler_secret)
if streamed:
result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": fail},
)
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,
"go",
error_handlers={"invalid_final_output": fail},
)
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
_assert_secret_absent_from_agents_traceback(error, _MODEL_OUTPUT_SECRET)
_assert_secret_absent_from_agents_traceback(error, handler_secret)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.asyncio
async def test_invalid_final_output_handler_failure_preserves_diagnostic_context(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", False)
model = ScriptedModel(steps=[[get_text_message(f'{{"answer": "{_MODEL_OUTPUT_SECRET}"}}')]])
agent = Agent(name="A", model=model, output_type=_RequiredOutput)
def fail(_data: RunErrorHandlerInput[None]) -> None:
raise RuntimeError("handler failed")
if streamed:
result = Runner.run_streamed(
agent,
"go",
error_handlers={"invalid_final_output": fail},
)
with pytest.raises(RuntimeError, match="handler failed") as exc_info:
async for _ in result.stream_events():
pass
else:
with pytest.raises(RuntimeError, match="handler failed") as exc_info:
await Runner.run(
agent,
"go",
error_handlers={"invalid_final_output": fail},
)
validation_error = exc_info.value.__context__
assert isinstance(validation_error, ModelBehaviorError)
assert _MODEL_OUTPUT_SECRET in str(validation_error)
assert isinstance(validation_error.__cause__, ValidationError)
@pytest.mark.parametrize("streamed", [False, True])
@pytest.mark.parametrize("redacted", [False, True])
@pytest.mark.asyncio
async def test_multiturn_output_validation_error_run_data_follows_redaction_policy(
monkeypatch: pytest.MonkeyPatch,
streamed: bool,
redacted: bool,
) -> None:
monkeypatch.setattr(_debug, "DONT_LOG_MODEL_DATA", redacted)
monkeypatch.setattr(_debug, "DONT_LOG_TOOL_DATA", redacted)
@function_tool
def record_value(value: str) -> str:
return "recorded"
model = ScriptedModel()
model.extend(
[
[
get_function_tool_call(
record_value.name,
json.dumps({"value": _MODEL_OUTPUT_SECRET}),
)
],
[get_text_message('{"answer": "missing count"}')],
]
)
agent = Agent(
name="A",
model=model,
tools=[record_value],
output_type=_RequiredOutput,
)
if streamed:
result = Runner.run_streamed(agent, "go")
with pytest.raises(ModelBehaviorError) as exc_info:
async for _ in result.stream_events():
pass
else:
with pytest.raises(ModelBehaviorError) as exc_info:
await Runner.run(agent, "go")
error = exc_info.value
if redacted:
assert error.run_data is None
assert _MODEL_OUTPUT_SECRET not in str(error)
_assert_secret_absent_from_agents_traceback(
error,
_MODEL_OUTPUT_SECRET,
require_agents_frames=streamed,
)
else:
assert error.run_data is not None
assert error.run_data.raw_responses
assert error.run_data.new_items
Reference in New Issue View Git Blame Copy Permalink