Kazuhiro Sera
2239 lines
70 KiB
Python
2239 lines
70 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any, cast
|
|
|
|
import httpx
|
|
import pytest
|
|
from openai import APIConnectionError
|
|
from openai.types.responses import (
|
|
ResponseApplyPatchToolCall,
|
|
ResponseCodeInterpreterToolCall,
|
|
ResponseCompletedEvent,
|
|
ResponseContentPartAddedEvent,
|
|
ResponseContentPartDoneEvent,
|
|
ResponseCreatedEvent,
|
|
ResponseCustomToolCall,
|
|
ResponseFunctionCallArgumentsDeltaEvent,
|
|
ResponseFunctionCallArgumentsDoneEvent,
|
|
ResponseFunctionToolCall,
|
|
ResponseInProgressEvent,
|
|
ResponseOutputItemAddedEvent,
|
|
ResponseOutputItemDoneEvent,
|
|
ResponseOutputMessage,
|
|
ResponseOutputRefusal,
|
|
ResponseOutputText,
|
|
ResponseOutputTextAnnotationAddedEvent,
|
|
ResponseReasoningSummaryPartAddedEvent,
|
|
ResponseReasoningTextDeltaEvent,
|
|
ResponseReasoningTextDoneEvent,
|
|
ResponseRefusalDeltaEvent,
|
|
ResponseRefusalDoneEvent,
|
|
ResponseTextDeltaEvent,
|
|
ResponseTextDoneEvent,
|
|
)
|
|
from openai.types.responses.response_output_item import ImageGenerationCall, McpCall
|
|
from openai.types.responses.response_output_text import (
|
|
AnnotationFilePath,
|
|
AnnotationURLCitation,
|
|
Logprob,
|
|
LogprobTopLogprob,
|
|
)
|
|
from openai.types.responses.response_prompt_param import ResponsePromptParam, Variables
|
|
from openai.types.responses.response_reasoning_item import Content, ResponseReasoningItem, Summary
|
|
from openai.types.responses.response_usage import InputTokensDetails, OutputTokensDetails
|
|
from pydantic import ValidationError
|
|
|
|
from agents import (
|
|
Agent,
|
|
ModelBehaviorError,
|
|
ModelRetryAdvice,
|
|
ModelRetryAdviceRequest,
|
|
ModelRetrySettings,
|
|
ModelTimeoutError,
|
|
RunConfig,
|
|
Runner,
|
|
handoff,
|
|
retry_policies,
|
|
)
|
|
from agents.agent_output import AgentOutputSchema
|
|
from agents.items import TResponseInputItem, TResponseOutputItem, TResponseStreamEvent
|
|
from agents.model_settings import ModelSettings
|
|
from agents.models.interface import ModelTracing
|
|
from agents.testing import (
|
|
InvalidModelStep,
|
|
ModelCall,
|
|
ModelStep,
|
|
ModelStepSpec,
|
|
ScriptedModel,
|
|
UnconsumedModelSteps,
|
|
UnexpectedModelCall,
|
|
assistant_message,
|
|
function_call,
|
|
)
|
|
from agents.tracing import SpanError
|
|
from agents.tracing.scope import Scope
|
|
from agents.usage import RequestUsage, Usage, _extract_raw_usage_snapshot
|
|
from agents.util._error_tracing import REDACTED_TRACE_ERROR_MESSAGE
|
|
|
|
from .model_test_helpers import get_response_obj
|
|
from .test_responses import get_function_tool
|
|
from .testing_processor import fetch_span_errors
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_runs_tool_workflow_and_records_calls() -> None:
|
|
model = ScriptedModel(
|
|
[
|
|
[function_call("lookup", {"city": "Tokyo"}, call_id="call_1")],
|
|
[assistant_message("sunny")],
|
|
]
|
|
)
|
|
agent = Agent(
|
|
name="weather",
|
|
model=model,
|
|
tools=[get_function_tool("lookup", "tool result")],
|
|
)
|
|
|
|
result = await Runner.run(agent, "weather?")
|
|
|
|
assert result.final_output == "sunny"
|
|
assert len(model.calls) == 2
|
|
assert model.first_call is not None
|
|
assert model.first_call.input == [{"content": "weather?", "role": "user"}]
|
|
assert model.last_call is not None
|
|
assert model.last_call.streamed is False
|
|
model.assert_complete()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_generates_stream_events() -> None:
|
|
model = ScriptedModel([[assistant_message("hello")]])
|
|
result = Runner.run_streamed(Agent(name="test", model=model), "hi")
|
|
|
|
events = [event async for event in result.stream_events()]
|
|
|
|
assert result.final_output == "hello"
|
|
assert events
|
|
assert model.last_call is not None
|
|
assert model.last_call.streamed is True
|
|
model.assert_complete()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_stream_preserves_request_usage_details() -> None:
|
|
request_entries = [
|
|
RequestUsage(
|
|
input_tokens=3,
|
|
output_tokens=2,
|
|
total_tokens=5,
|
|
input_tokens_details=InputTokensDetails.model_validate(
|
|
{"cached_tokens": 1, "cache_write_tokens": 0}
|
|
),
|
|
output_tokens_details=OutputTokensDetails(reasoning_tokens=1),
|
|
),
|
|
RequestUsage(
|
|
input_tokens=7,
|
|
output_tokens=4,
|
|
total_tokens=11,
|
|
input_tokens_details=InputTokensDetails.model_validate(
|
|
{"cached_tokens": 2, "cache_write_tokens": 3}
|
|
),
|
|
output_tokens_details=OutputTokensDetails(reasoning_tokens=2),
|
|
),
|
|
]
|
|
usage = Usage(
|
|
requests=5,
|
|
input_tokens=10,
|
|
output_tokens=6,
|
|
total_tokens=16,
|
|
input_tokens_details=InputTokensDetails.model_validate(
|
|
{"cached_tokens": 3, "cache_write_tokens": 3}
|
|
),
|
|
output_tokens_details=OutputTokensDetails(reasoning_tokens=3),
|
|
request_usage_entries=request_entries,
|
|
)
|
|
streamed_model = ScriptedModel([ModelStep(output=[assistant_message("hello")], usage=usage)])
|
|
non_streamed_model = ScriptedModel(
|
|
[ModelStep(output=[assistant_message("hello")], usage=usage)]
|
|
)
|
|
|
|
streamed_result = Runner.run_streamed(Agent(name="streamed", model=streamed_model), "hi")
|
|
async for _event in streamed_result.stream_events():
|
|
pass
|
|
non_streamed_result = await Runner.run(
|
|
Agent(name="non-streamed", model=non_streamed_model), "hi"
|
|
)
|
|
|
|
propagated = streamed_result.context_wrapper.usage
|
|
assert propagated.requests == 5
|
|
assert propagated.input_tokens == 10
|
|
assert propagated.output_tokens == 6
|
|
assert propagated.total_tokens == 16
|
|
assert propagated.request_usage_entries == request_entries
|
|
assert propagated == non_streamed_result.context_wrapper.usage
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_stream_counts_default_usage_as_one_request() -> None:
|
|
streamed_model = ScriptedModel([[assistant_message("streamed")]])
|
|
non_streamed_model = ScriptedModel([[assistant_message("non-streamed")]])
|
|
|
|
streamed_result = Runner.run_streamed(Agent(name="streamed", model=streamed_model), "hi")
|
|
async for _event in streamed_result.stream_events():
|
|
pass
|
|
non_streamed_result = await Runner.run(
|
|
Agent(name="non-streamed", model=non_streamed_model), "hi"
|
|
)
|
|
|
|
assert streamed_result.context_wrapper.usage.requests == 1
|
|
assert non_streamed_result.context_wrapper.usage.requests == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_preserves_explicit_zero_request_usage_across_run_modes() -> None:
|
|
usage = Usage(
|
|
input_tokens=2,
|
|
output_tokens=1,
|
|
total_tokens=3,
|
|
request_usage_entries=[
|
|
RequestUsage(
|
|
input_tokens=1,
|
|
output_tokens=1,
|
|
total_tokens=2,
|
|
input_tokens_details=InputTokensDetails.model_validate(
|
|
{"cached_tokens": 0, "cache_write_tokens": 0}
|
|
),
|
|
output_tokens_details=OutputTokensDetails(reasoning_tokens=0),
|
|
)
|
|
],
|
|
)
|
|
streamed_model = ScriptedModel([ModelStep(output=[assistant_message("streamed")], usage=usage)])
|
|
non_streamed_model = ScriptedModel(
|
|
[ModelStep(output=[assistant_message("non-streamed")], usage=usage)]
|
|
)
|
|
|
|
streamed_result = Runner.run_streamed(Agent(name="streamed", model=streamed_model), "hi")
|
|
async for _event in streamed_result.stream_events():
|
|
pass
|
|
non_streamed_result = await Runner.run(
|
|
Agent(name="non-streamed", model=non_streamed_model), "hi"
|
|
)
|
|
|
|
assert streamed_result.context_wrapper.usage == usage
|
|
assert non_streamed_result.context_wrapper.usage == usage
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_copies_default_usage_before_retry_accounting() -> None:
|
|
default_usage = Usage(requests=1, input_tokens=2, total_tokens=2)
|
|
model = ScriptedModel(
|
|
[
|
|
APIConnectionError(
|
|
message="connection error",
|
|
request=httpx.Request("POST", "https://example.com"),
|
|
),
|
|
[assistant_message("first")],
|
|
[assistant_message("second")],
|
|
],
|
|
default_usage=default_usage,
|
|
)
|
|
agent = Agent(
|
|
name="test",
|
|
model=model,
|
|
model_settings=ModelSettings(
|
|
retry=ModelRetrySettings(
|
|
max_retries=1,
|
|
policy=retry_policies.network_error(),
|
|
)
|
|
),
|
|
)
|
|
|
|
first_result = await Runner.run(agent, "first")
|
|
second_result = await Runner.run(agent, "second")
|
|
|
|
assert first_result.context_wrapper.usage.requests == 2
|
|
assert second_result.context_wrapper.usage.requests == 1
|
|
assert default_usage == Usage(requests=1, input_tokens=2, total_tokens=2)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("response_id", [None, ""])
|
|
async def test_scripted_model_stream_preserves_response_id(response_id: str | None) -> None:
|
|
streamed_model = ScriptedModel(
|
|
[ModelStep(output=[assistant_message("streamed")], response_id=response_id)]
|
|
)
|
|
non_streamed_model = ScriptedModel(
|
|
[ModelStep(output=[assistant_message("non-streamed")], response_id=response_id)]
|
|
)
|
|
|
|
streamed_result = Runner.run_streamed(Agent(name="streamed", model=streamed_model), "hi")
|
|
async for _event in streamed_result.stream_events():
|
|
pass
|
|
non_streamed_result = await Runner.run(
|
|
Agent(name="non-streamed", model=non_streamed_model), "hi"
|
|
)
|
|
|
|
assert streamed_result.last_response_id == response_id
|
|
assert non_streamed_result.last_response_id == response_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_stream_does_not_chain_absent_response_id() -> None:
|
|
def respond(call: ModelCall):
|
|
assert call.previous_response_id is None
|
|
return [assistant_message("done")]
|
|
|
|
model = ScriptedModel(
|
|
[
|
|
ModelStep(
|
|
output=[function_call("lookup", {}, call_id="call_1")],
|
|
response_id=None,
|
|
),
|
|
ModelStep.respond(respond),
|
|
]
|
|
)
|
|
agent = Agent(
|
|
name="test",
|
|
model=model,
|
|
tools=[get_function_tool("lookup", "tool result")],
|
|
)
|
|
|
|
result = Runner.run_streamed(agent, "hi", auto_previous_response_id=True)
|
|
async for _event in result.stream_events():
|
|
pass
|
|
|
|
assert result.final_output == "done"
|
|
model.assert_complete()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_supports_dynamic_responder() -> None:
|
|
def respond(call):
|
|
assert isinstance(call.input, list)
|
|
return [assistant_message(str(call.input[0]["content"]))]
|
|
|
|
model = ScriptedModel([ModelStep.respond(respond)])
|
|
|
|
result = await Runner.run(Agent(name="test", model=model), "hello")
|
|
|
|
assert result.final_output == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_accepts_step_mapping() -> None:
|
|
usage = Usage(requests=2, input_tokens=3, output_tokens=4, total_tokens=7)
|
|
model = ScriptedModel(
|
|
[
|
|
{
|
|
"output": [assistant_message("mapped")],
|
|
"usage": usage,
|
|
"response_id": "resp_mapped",
|
|
"request_id": "req_mapped",
|
|
"raw_usage": {"source": "mapping"},
|
|
}
|
|
]
|
|
)
|
|
|
|
response = await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
assert response.output == [assistant_message("mapped")]
|
|
assert response.usage == usage
|
|
assert response.response_id == "resp_mapped"
|
|
assert response.request_id == "req_mapped"
|
|
assert response.raw_usage is None
|
|
model.assert_complete()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_snapshots_model_settings_in_recorded_calls() -> None:
|
|
settings = ModelSettings(
|
|
tool_choice="auto",
|
|
extra_args={"provider": {"mode": "first"}},
|
|
)
|
|
model = ScriptedModel([[assistant_message("first")], [assistant_message("second")]])
|
|
|
|
async for _event in model.stream_response(
|
|
None,
|
|
[],
|
|
settings,
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
pass
|
|
settings.tool_choice = "none"
|
|
assert settings.extra_args is not None
|
|
settings.extra_args["provider"]["mode"] = "second"
|
|
await model.get_response(
|
|
None,
|
|
[],
|
|
settings,
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
settings.extra_args["provider"]["mode"] = "after"
|
|
|
|
first_settings = model.calls[0].model_settings
|
|
second_settings = model.calls[1].model_settings
|
|
assert first_settings is not settings
|
|
assert first_settings.tool_choice == "auto"
|
|
assert first_settings.extra_args == {"provider": {"mode": "first"}}
|
|
assert second_settings.tool_choice == "none"
|
|
assert second_settings.extra_args == {"provider": {"mode": "second"}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_snapshots_input_and_prompt_in_recorded_calls() -> None:
|
|
input_item: dict[str, Any] = {
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": "first"}],
|
|
}
|
|
input_items = cast(list[TResponseInputItem], [input_item])
|
|
prompt_variables: dict[str, Variables] = {"topic": "first"}
|
|
prompt: ResponsePromptParam = {
|
|
"id": "pmpt_1",
|
|
"variables": prompt_variables,
|
|
}
|
|
model = ScriptedModel([[assistant_message("done")]])
|
|
|
|
await model.get_response(
|
|
None,
|
|
input_items,
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=prompt,
|
|
)
|
|
input_item["content"][0]["text"] = "second"
|
|
prompt_variables["topic"] = "second"
|
|
|
|
assert model.last_call is not None
|
|
assert model.last_call.input == [
|
|
{
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": "first"}],
|
|
}
|
|
]
|
|
assert model.last_call.prompt == {
|
|
"id": "pmpt_1",
|
|
"variables": {"topic": "first"},
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_exposes_detached_read_only_call_history() -> None:
|
|
input_item: dict[str, Any] = {
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": "first"}],
|
|
}
|
|
settings = ModelSettings(extra_args={"provider": {"mode": "first"}})
|
|
tool = get_function_tool("lookup", "tool result")
|
|
target = Agent(name="delegate")
|
|
handoff_value = handoff(target)
|
|
output_schema = AgentOutputSchema(str)
|
|
model = ScriptedModel([[assistant_message("done")]])
|
|
|
|
await model.get_response(
|
|
None,
|
|
cast(list[TResponseInputItem], [input_item]),
|
|
settings,
|
|
[tool],
|
|
output_schema,
|
|
[handoff_value],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
history = model.calls
|
|
assert isinstance(history, tuple)
|
|
with pytest.raises(AttributeError):
|
|
cast(Any, history).append(history[0])
|
|
|
|
returned = history[0]
|
|
returned.input[0]["content"][0]["text"] = "changed"
|
|
assert returned.model_settings.extra_args is not None
|
|
returned.model_settings.extra_args["provider"]["mode"] = "changed"
|
|
returned.tools.clear()
|
|
returned.handoffs.clear()
|
|
|
|
retained = model.calls[0]
|
|
assert retained.input[0]["content"][0]["text"] == "first"
|
|
assert retained.model_settings.extra_args == {"provider": {"mode": "first"}}
|
|
assert retained.tools == [tool]
|
|
assert retained.handoffs == [handoff_value]
|
|
assert retained.tools[0] is tool
|
|
assert retained.handoffs[0] is handoff_value
|
|
assert retained.output_schema is output_schema
|
|
assert retained.tracing is ModelTracing.DISABLED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
async def test_scripted_model_snapshot_failure_has_no_recording_side_effects(
|
|
streamed: bool,
|
|
) -> None:
|
|
expected = RuntimeError("settings snapshot failed")
|
|
|
|
class Uncopyable:
|
|
def __deepcopy__(self, _memo: dict[int, Any]) -> Any:
|
|
raise expected
|
|
|
|
responder_calls: list[ModelCall] = []
|
|
|
|
def respond(call: ModelCall) -> list[TResponseOutputItem]:
|
|
responder_calls.append(call)
|
|
return [assistant_message("unexpected")]
|
|
|
|
model = ScriptedModel([ModelStep.respond(respond)])
|
|
settings = ModelSettings(extra_args={"sentinel": Uncopyable()})
|
|
|
|
with pytest.raises(RuntimeError, match="settings snapshot failed") as exc_info:
|
|
if streamed:
|
|
async for _event in model.stream_response(
|
|
None,
|
|
[],
|
|
settings,
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
pass
|
|
else:
|
|
await model.get_response(
|
|
None,
|
|
[],
|
|
settings,
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
assert exc_info.value is expected
|
|
assert model.calls == ()
|
|
assert model.remaining_steps == 1
|
|
assert responder_calls == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
@pytest.mark.parametrize("preserve_raw_usage", [None, False, True])
|
|
async def test_scripted_model_gates_and_snapshots_raw_usage(
|
|
streamed: bool,
|
|
preserve_raw_usage: bool | None,
|
|
) -> None:
|
|
raw_usage = {"provider": {"input_tokens": 3}}
|
|
model = ScriptedModel([ModelStep(output=[assistant_message("done")], raw_usage=raw_usage)])
|
|
agent = Agent(
|
|
name="test",
|
|
model=model,
|
|
model_settings=ModelSettings(preserve_raw_usage=preserve_raw_usage),
|
|
)
|
|
|
|
if streamed:
|
|
result = Runner.run_streamed(agent, "hi")
|
|
terminal_raw_usage = None
|
|
saw_terminal = False
|
|
async for event in result.stream_events():
|
|
if event.type == "raw_response_event" and isinstance(
|
|
event.data, ResponseCompletedEvent
|
|
):
|
|
saw_terminal = True
|
|
terminal_raw_usage = _extract_raw_usage_snapshot(event.data.response)
|
|
assert saw_terminal is True
|
|
else:
|
|
result = await Runner.run(agent, "hi")
|
|
terminal_raw_usage = None
|
|
|
|
raw_usage["provider"]["input_tokens"] = 99
|
|
expected_raw_usage = {"provider": {"input_tokens": 3}} if preserve_raw_usage is True else None
|
|
assert result.raw_responses[0].raw_usage == expected_raw_usage
|
|
if streamed:
|
|
assert terminal_raw_usage == expected_raw_usage
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_withholds_raw_usage_until_stream_completion() -> None:
|
|
raw_usage = {"provider": {"input_tokens": 3}}
|
|
model = ScriptedModel([ModelStep(output=[assistant_message("done")], raw_usage=raw_usage)])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(preserve_raw_usage=True),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
response_events = [
|
|
event
|
|
for event in events
|
|
if isinstance(
|
|
event,
|
|
ResponseCreatedEvent | ResponseInProgressEvent | ResponseCompletedEvent,
|
|
)
|
|
]
|
|
assert [_extract_raw_usage_snapshot(event.response) for event in response_events] == [
|
|
None,
|
|
None,
|
|
raw_usage,
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_accepts_step_mappings_in_queue_methods() -> None:
|
|
model = ScriptedModel()
|
|
model.enqueue({"output": [assistant_message("first")]})
|
|
model.extend([{"output": [assistant_message("second")]}])
|
|
agent = Agent(name="test", model=model)
|
|
|
|
first = await Runner.run(agent, "first")
|
|
second = await Runner.run(agent, "second")
|
|
|
|
assert first.final_output == "first"
|
|
assert second.final_output == "second"
|
|
model.assert_complete()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_responder_can_return_step_mapping() -> None:
|
|
def respond(call: ModelCall) -> ModelStepSpec:
|
|
assert isinstance(call.input, list)
|
|
return {"output": [assistant_message(str(call.input[0]["content"]))]}
|
|
|
|
model = ScriptedModel([ModelStep.respond(respond)])
|
|
|
|
result = await Runner.run(Agent(name="test", model=model), "mapped")
|
|
|
|
assert result.final_output == "mapped"
|
|
model.assert_complete()
|
|
|
|
|
|
def test_scripted_model_rejects_unknown_step_mapping_keys() -> None:
|
|
invalid_step = cast(ModelStepSpec, {"unknown": True})
|
|
|
|
with pytest.raises(InvalidModelStep, match=r"step #1") as exc_info:
|
|
ScriptedModel([invalid_step])
|
|
|
|
assert exc_info.value.reason == "unsupported_field"
|
|
assert exc_info.value.input_index == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("step", "reason"),
|
|
[
|
|
({"error": "not an exception"}, "invalid_error"),
|
|
({"responder": "not callable"}, "invalid_responder"),
|
|
({"stream_events": 1}, "invalid_stream_events"),
|
|
(
|
|
{"error": RuntimeError("failed"), "responder": lambda _call: []},
|
|
"conflicting_outcomes",
|
|
),
|
|
({"output": "not an output sequence"}, "invalid_input"),
|
|
({"usage": {"requests": 1}}, "invalid_input"),
|
|
({"response_id": 1}, "invalid_input"),
|
|
({"raw_usage": []}, "invalid_input"),
|
|
({"retry_advice": ModelRetryAdvice(suggested=True)}, "invalid_retry_advice"),
|
|
(
|
|
{"error": RuntimeError("failed"), "retry_advice": "not advice"},
|
|
"invalid_retry_advice",
|
|
),
|
|
],
|
|
)
|
|
def test_scripted_model_validates_step_envelopes_before_queuing(
|
|
step: object,
|
|
reason: str,
|
|
) -> None:
|
|
model = ScriptedModel([[assistant_message("retained")]])
|
|
|
|
with pytest.raises(InvalidModelStep, match=r"step #1") as exc_info:
|
|
model.extend(cast(Any, [step]))
|
|
|
|
assert exc_info.value.reason == reason
|
|
assert exc_info.value.input_index == 0
|
|
assert model.remaining_steps == 1
|
|
|
|
|
|
def test_scripted_model_reports_the_zero_origin_index_of_an_invalid_step() -> None:
|
|
with pytest.raises(InvalidModelStep, match=r"step #2") as exc_info:
|
|
ScriptedModel([[assistant_message("valid")], cast(Any, {"unknown": True})])
|
|
|
|
assert exc_info.value.reason == "unsupported_field"
|
|
assert exc_info.value.input_index == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_supports_awaitable_recursive_responder() -> None:
|
|
async def outer_responder(call: ModelCall) -> ModelStep:
|
|
def inner_responder(inner_call: ModelCall) -> list[TResponseOutputItem]:
|
|
assert inner_call is call
|
|
return [assistant_message("nested")]
|
|
|
|
return ModelStep.respond(inner_responder)
|
|
|
|
model = ScriptedModel([ModelStep.respond(outer_responder)])
|
|
|
|
result = await Runner.run(Agent(name="test", model=model), "hello")
|
|
|
|
assert result.final_output == "nested"
|
|
assert len(model.calls) == 1
|
|
model.assert_complete()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_retry_advice_is_error_scoped_and_detached() -> None:
|
|
error = RuntimeError("failed")
|
|
advice = ModelRetryAdvice(suggested=True, replay_safety="safe")
|
|
model = ScriptedModel([ModelStep.raise_error(error, retry_advice=advice)])
|
|
advice.suggested = False
|
|
|
|
with pytest.raises(RuntimeError, match="failed"):
|
|
await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
request = ModelRetryAdviceRequest(error=error, attempt=1, stream=False)
|
|
first = model.get_retry_advice(request)
|
|
assert first is not None
|
|
assert first.suggested is True
|
|
first.suggested = False
|
|
second = model.get_retry_advice(request)
|
|
assert second is not None
|
|
assert second.suggested is True
|
|
other_request = ModelRetryAdviceRequest(
|
|
error=RuntimeError("other"),
|
|
attempt=1,
|
|
stream=False,
|
|
)
|
|
assert model.get_retry_advice(other_request) is None
|
|
|
|
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_clears_stale_retry_advice_after_responder_failure(
|
|
streamed: bool,
|
|
) -> None:
|
|
error = RuntimeError("failed")
|
|
|
|
def raise_same_error(_call: ModelCall) -> ModelStep:
|
|
raise error
|
|
|
|
model = ScriptedModel(
|
|
[
|
|
ModelStep.raise_error(error, retry_advice=ModelRetryAdvice(suggested=True)),
|
|
ModelStep.respond(raise_same_error),
|
|
[assistant_message("must remain unconsumed")],
|
|
]
|
|
)
|
|
|
|
async def invoke() -> None:
|
|
if streamed:
|
|
async for _event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
pass
|
|
else:
|
|
await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
request = ModelRetryAdviceRequest(error=error, attempt=1, stream=streamed)
|
|
with pytest.raises(RuntimeError, match="failed"):
|
|
await invoke()
|
|
assert model.get_retry_advice(request) is not None
|
|
|
|
with pytest.raises(RuntimeError, match="failed"):
|
|
await invoke()
|
|
assert model.get_retry_advice(request) is None
|
|
assert model.remaining_steps == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_supports_exact_normalized_stream() -> None:
|
|
output = [assistant_message("exact")]
|
|
|
|
async def stream(_call) -> AsyncIterator[TResponseStreamEvent]:
|
|
yield ResponseCompletedEvent(
|
|
type="response.completed",
|
|
response=get_response_obj(output),
|
|
sequence_number=0,
|
|
)
|
|
|
|
model = ScriptedModel([ModelStep.stream(stream, output=output)])
|
|
result = Runner.run_streamed(Agent(name="test", model=model), "hi")
|
|
|
|
async for _event in result.stream_events():
|
|
pass
|
|
|
|
assert result.final_output == "exact"
|
|
|
|
|
|
def test_model_step_freezes_exact_stream_event_sequence() -> None:
|
|
event = ResponseCompletedEvent(
|
|
type="response.completed",
|
|
response=get_response_obj([]),
|
|
sequence_number=0,
|
|
)
|
|
events = [event]
|
|
|
|
step = ModelStep.stream(events)
|
|
events.clear()
|
|
|
|
assert step.stream_events == (event,)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("output_item", "item_kind"),
|
|
[
|
|
(
|
|
ResponseCustomToolCall(
|
|
call_id="call_1",
|
|
id="item_1",
|
|
input="payload",
|
|
name="custom",
|
|
type="custom_tool_call",
|
|
),
|
|
"ResponseCustomToolCall",
|
|
),
|
|
(
|
|
ResponseCodeInterpreterToolCall(
|
|
id="item_1",
|
|
code="print('hello')",
|
|
container_id="container_1",
|
|
outputs=None,
|
|
status="completed",
|
|
type="code_interpreter_call",
|
|
),
|
|
"ResponseCodeInterpreterToolCall",
|
|
),
|
|
(
|
|
McpCall(
|
|
id="item_1",
|
|
arguments='{"query":"hello"}',
|
|
name="search",
|
|
server_label="docs",
|
|
status="completed",
|
|
type="mcp_call",
|
|
),
|
|
"McpCall",
|
|
),
|
|
(
|
|
ImageGenerationCall(
|
|
id="item_1",
|
|
result="base64-image",
|
|
status="completed",
|
|
type="image_generation_call",
|
|
),
|
|
"ImageGenerationCall",
|
|
),
|
|
],
|
|
)
|
|
async def test_scripted_model_rejects_unsupported_automatic_tool_streams(
|
|
output_item: TResponseOutputItem,
|
|
item_kind: str,
|
|
) -> None:
|
|
model = ScriptedModel([[output_item]])
|
|
yielded: list[TResponseStreamEvent] = []
|
|
|
|
with pytest.raises(
|
|
ModelBehaviorError,
|
|
match=rf"Automatic streaming does not support {item_kind}.*ModelStep\.stream",
|
|
):
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
yielded.append(event)
|
|
|
|
assert yielded == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("queue_method", ["constructor", "enqueue", "extend"])
|
|
async def test_scripted_model_snapshots_static_steps_when_queued(queue_method: str) -> None:
|
|
output_item = assistant_message("before")
|
|
usage = Usage(requests=2, input_tokens=3, output_tokens=4, total_tokens=7)
|
|
raw_usage = {"provider": {"cached_tokens": 1}}
|
|
step = ModelStep(
|
|
output=[output_item],
|
|
usage=usage,
|
|
response_id="resp_before",
|
|
raw_usage=raw_usage,
|
|
)
|
|
|
|
if queue_method == "constructor":
|
|
model = ScriptedModel([step])
|
|
else:
|
|
model = ScriptedModel()
|
|
if queue_method == "enqueue":
|
|
model.enqueue(step)
|
|
else:
|
|
model.extend([step])
|
|
|
|
message = cast(ResponseOutputMessage, output_item)
|
|
text = cast(ResponseOutputText, message.content[0])
|
|
text.text = "after"
|
|
usage.requests = 99
|
|
raw_usage["provider"]["cached_tokens"] = 99
|
|
step.response_id = "resp_after"
|
|
|
|
response = await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(preserve_raw_usage=True),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
queued_message = cast(ResponseOutputMessage, response.output[0])
|
|
queued_text = cast(ResponseOutputText, queued_message.content[0])
|
|
assert queued_text.text == "before"
|
|
assert response.usage == Usage(requests=2, input_tokens=3, output_tokens=4, total_tokens=7)
|
|
assert response.response_id == "resp_before"
|
|
assert response.raw_usage == {"provider": {"cached_tokens": 1}}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_snapshots_output_shorthand_when_queued() -> None:
|
|
output_item = assistant_message("before")
|
|
output = [output_item]
|
|
model = ScriptedModel([output])
|
|
|
|
message = cast(ResponseOutputMessage, output_item)
|
|
text = cast(ResponseOutputText, message.content[0])
|
|
text.text = "after"
|
|
output.clear()
|
|
|
|
response = await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
queued_message = cast(ResponseOutputMessage, response.output[0])
|
|
queued_text = cast(ResponseOutputText, queued_message.content[0])
|
|
assert queued_text.text == "before"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_records_stream_factory_errors_on_generation_span() -> None:
|
|
async def stream(_call) -> AsyncIterator[TResponseStreamEvent]:
|
|
yield ResponseCreatedEvent(
|
|
type="response.created",
|
|
response=get_response_obj([]),
|
|
sequence_number=0,
|
|
)
|
|
raise RuntimeError("stream failed")
|
|
|
|
model = ScriptedModel([ModelStep.stream(stream)], emit_traces=True)
|
|
|
|
with pytest.raises(RuntimeError, match="stream failed"):
|
|
result = Runner.run_streamed(Agent(name="test", model=model), "hi")
|
|
async for _event in result.stream_events():
|
|
pass
|
|
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {"name": "RuntimeError", "message": "stream failed"},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_marks_span_current_only_while_advancing_stream_factory() -> None:
|
|
observed_spans: list[Any] = []
|
|
|
|
async def stream(_call) -> AsyncIterator[TResponseStreamEvent]:
|
|
observed_spans.append(Scope.get_current_span())
|
|
yield ResponseCreatedEvent(
|
|
type="response.created",
|
|
response=get_response_obj([]),
|
|
sequence_number=0,
|
|
)
|
|
observed_spans.append(Scope.get_current_span())
|
|
|
|
model = ScriptedModel([ModelStep.stream(stream)], emit_traces=True)
|
|
|
|
async for _event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
assert Scope.get_current_span() is None
|
|
|
|
assert len(observed_spans) == 2
|
|
assert observed_spans[0] is observed_spans[1]
|
|
assert observed_spans[0] is not None
|
|
assert observed_spans[0].span_data.type == "generation"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_closes_exact_stream_when_outer_stream_is_closed() -> None:
|
|
closed = False
|
|
close_span: Any = None
|
|
|
|
async def stream(_call) -> AsyncIterator[TResponseStreamEvent]:
|
|
nonlocal closed, close_span
|
|
try:
|
|
yield ResponseCreatedEvent(
|
|
type="response.created",
|
|
response=get_response_obj([]),
|
|
sequence_number=0,
|
|
)
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
closed = True
|
|
close_span = Scope.get_current_span()
|
|
|
|
model = ScriptedModel([ModelStep.stream(stream)], emit_traces=True)
|
|
outer = model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
await anext(outer)
|
|
assert closed is False
|
|
await outer.aclose()
|
|
|
|
assert closed is True
|
|
assert close_span is not None
|
|
assert close_span.span_data.type == "generation"
|
|
assert Scope.get_current_span() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_closes_exact_stream_when_consumer_is_cancelled() -> None:
|
|
blocked = asyncio.Event()
|
|
closed = asyncio.Event()
|
|
|
|
async def stream(_call) -> AsyncIterator[TResponseStreamEvent]:
|
|
try:
|
|
yield ResponseCreatedEvent(
|
|
type="response.created",
|
|
response=get_response_obj([]),
|
|
sequence_number=0,
|
|
)
|
|
blocked.set()
|
|
await asyncio.Event().wait()
|
|
finally:
|
|
closed.set()
|
|
|
|
model = ScriptedModel([ModelStep.stream(stream)])
|
|
outer = model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
await anext(outer)
|
|
pending = asyncio.create_task(anext(outer))
|
|
await blocked.wait()
|
|
pending.cancel()
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await pending
|
|
assert closed.is_set()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_preserves_cancellation_when_exact_stream_close_fails() -> None:
|
|
class FailingCloseStream:
|
|
def __init__(self) -> None:
|
|
self.blocked = asyncio.Event()
|
|
self.close_calls = 0
|
|
|
|
def __aiter__(self) -> FailingCloseStream:
|
|
return self
|
|
|
|
async def __anext__(self) -> TResponseStreamEvent:
|
|
self.blocked.set()
|
|
await asyncio.Event().wait()
|
|
raise AssertionError("unreachable")
|
|
|
|
async def aclose(self) -> None:
|
|
self.close_calls += 1
|
|
raise RuntimeError("close failed")
|
|
|
|
inner = FailingCloseStream()
|
|
model = ScriptedModel([ModelStep.stream(lambda _call: inner)])
|
|
outer = model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
pending = asyncio.create_task(anext(outer))
|
|
await inner.blocked.wait()
|
|
pending.cancel()
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await pending
|
|
assert inner.close_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_surfaces_exact_stream_close_failure_after_exhaustion() -> None:
|
|
close_error = RuntimeError("close failed")
|
|
|
|
class FailingCloseStream:
|
|
def __aiter__(self) -> FailingCloseStream:
|
|
return self
|
|
|
|
async def __anext__(self) -> TResponseStreamEvent:
|
|
raise StopAsyncIteration
|
|
|
|
async def aclose(self) -> None:
|
|
raise close_error
|
|
|
|
inner = FailingCloseStream()
|
|
model = ScriptedModel([ModelStep.stream(lambda _call: inner)], emit_traces=True)
|
|
agent = Agent(name="test", model=model)
|
|
|
|
with pytest.raises(RuntimeError, match="close failed") as exc_info:
|
|
result = Runner.run_streamed(agent, "hi")
|
|
async for _event in result.stream_events():
|
|
pass
|
|
|
|
assert exc_info.value is close_error
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {"name": "RuntimeError", "message": "close failed"},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_preserves_exact_stream_error_when_close_fails() -> None:
|
|
stream_error = RuntimeError("stream failed")
|
|
|
|
class FailingStream:
|
|
def __init__(self) -> None:
|
|
self.close_calls = 0
|
|
|
|
def __aiter__(self) -> FailingStream:
|
|
return self
|
|
|
|
async def __anext__(self) -> TResponseStreamEvent:
|
|
raise stream_error
|
|
|
|
async def aclose(self) -> None:
|
|
self.close_calls += 1
|
|
raise RuntimeError("close failed")
|
|
|
|
inner = FailingStream()
|
|
model = ScriptedModel([ModelStep.stream(lambda _call: inner)], emit_traces=True)
|
|
agent = Agent(name="test", model=model)
|
|
|
|
with pytest.raises(RuntimeError, match="stream failed") as exc_info:
|
|
result = Runner.run_streamed(agent, "hi")
|
|
async for _event in result.stream_events():
|
|
pass
|
|
|
|
assert exc_info.value is stream_error
|
|
assert inner.close_calls == 1
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {"name": "RuntimeError", "message": "stream failed"},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("emit_traces", [False, True])
|
|
def test_scripted_model_early_stream_exit_has_task_safe_span_cleanup(emit_traces: bool) -> None:
|
|
loop = asyncio.new_event_loop()
|
|
errors: list[dict[str, Any]] = []
|
|
loop.set_exception_handler(lambda _loop, context: errors.append(context))
|
|
|
|
async def consume_one_event() -> None:
|
|
model = ScriptedModel([[assistant_message("hello")]], emit_traces=emit_traces)
|
|
async for _event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
assert Scope.get_current_span() is None
|
|
break
|
|
|
|
try:
|
|
loop.run_until_complete(consume_one_event())
|
|
loop.run_until_complete(loop.shutdown_asyncgens())
|
|
finally:
|
|
loop.close()
|
|
|
|
assert errors == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
async def test_scripted_model_preserves_unformattable_responder_error(streamed: bool) -> None:
|
|
class UnformattableError(RuntimeError):
|
|
def __str__(self) -> str:
|
|
raise RuntimeError("format failed")
|
|
|
|
expected = UnformattableError()
|
|
|
|
def respond(_call: ModelCall) -> Any:
|
|
raise expected
|
|
|
|
model = ScriptedModel([ModelStep.respond(respond)], emit_traces=True)
|
|
agent = Agent(name="test", model=model)
|
|
|
|
with pytest.raises(UnformattableError) as exc_info:
|
|
if streamed:
|
|
result = Runner.run_streamed(agent, "hi")
|
|
async for _event in result.stream_events():
|
|
pass
|
|
else:
|
|
await Runner.run(agent, "hi")
|
|
|
|
assert exc_info.value is expected
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {
|
|
"name": "UnformattableError",
|
|
"message": "Unrenderable UnformattableError",
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
async def test_scripted_model_timeout_records_generation_span_error(streamed: bool) -> None:
|
|
async def respond(_call: ModelCall) -> Any:
|
|
await asyncio.Event().wait()
|
|
raise AssertionError("unreachable")
|
|
|
|
model = ScriptedModel([ModelStep.respond(respond)], emit_traces=True)
|
|
agent = Agent(
|
|
name="test",
|
|
model=model,
|
|
model_settings=ModelSettings(timeout=0.01),
|
|
)
|
|
|
|
with pytest.raises(ModelTimeoutError):
|
|
if streamed:
|
|
result = Runner.run_streamed(agent, "hi")
|
|
async for _event in result.stream_events():
|
|
pass
|
|
else:
|
|
await Runner.run(agent, "hi")
|
|
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {"error": "Model call timed out after 0.01 seconds."},
|
|
}
|
|
]
|
|
|
|
|
|
def test_scripted_model_ignores_span_attachment_failure() -> None:
|
|
class FailingSpan:
|
|
def set_error(self, _error: SpanError) -> None:
|
|
raise KeyboardInterrupt
|
|
|
|
ScriptedModel._set_span_error(FailingSpan(), RuntimeError("model failed"), ModelTracing.ENABLED)
|
|
|
|
|
|
def test_scripted_model_contains_base_exception_from_error_formatting() -> None:
|
|
class UnformattableError(RuntimeError):
|
|
def __str__(self) -> str:
|
|
raise KeyboardInterrupt
|
|
|
|
class RecordingSpan:
|
|
def __init__(self) -> None:
|
|
self.error: SpanError | None = None
|
|
|
|
def set_error(self, error: SpanError) -> None:
|
|
self.error = error
|
|
|
|
span = RecordingSpan()
|
|
ScriptedModel._set_span_error(span, UnformattableError(), ModelTracing.ENABLED)
|
|
|
|
assert span.error == SpanError(
|
|
message="Error",
|
|
data={
|
|
"name": "UnformattableError",
|
|
"message": "Unrenderable UnformattableError",
|
|
},
|
|
)
|
|
|
|
|
|
def test_scripted_model_redacts_span_error_without_rendering_exception() -> None:
|
|
class SensitiveError(RuntimeError):
|
|
def __init__(self) -> None:
|
|
self.str_calls = 0
|
|
|
|
def __str__(self) -> str:
|
|
self.str_calls += 1
|
|
return "sensitive payload"
|
|
|
|
class RecordingSpan:
|
|
def __init__(self) -> None:
|
|
self.error: SpanError | None = None
|
|
|
|
def set_error(self, error: SpanError) -> None:
|
|
self.error = error
|
|
|
|
error = SensitiveError()
|
|
span = RecordingSpan()
|
|
|
|
ScriptedModel._set_span_error(span, error, ModelTracing.ENABLED_WITHOUT_DATA)
|
|
|
|
assert error.str_calls == 0
|
|
assert span.error == SpanError(
|
|
message="Error",
|
|
data={
|
|
"name": "SensitiveError",
|
|
"message": REDACTED_TRACE_ERROR_MESSAGE,
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
@pytest.mark.parametrize("awaitable", [False, True])
|
|
async def test_scripted_model_records_responder_errors_on_generation_span(
|
|
streamed: bool,
|
|
awaitable: bool,
|
|
) -> None:
|
|
expected = RuntimeError("responder failed")
|
|
|
|
def respond(_call: ModelCall) -> Any:
|
|
raise expected
|
|
|
|
async def respond_async(_call: ModelCall) -> Any:
|
|
raise expected
|
|
|
|
model = ScriptedModel(
|
|
[ModelStep.respond(respond_async if awaitable else respond)],
|
|
emit_traces=True,
|
|
)
|
|
agent = Agent(name="test", model=model)
|
|
|
|
with pytest.raises(RuntimeError, match="responder failed") as exc_info:
|
|
if streamed:
|
|
result = Runner.run_streamed(agent, "hi")
|
|
async for _event in result.stream_events():
|
|
pass
|
|
else:
|
|
await Runner.run(agent, "hi")
|
|
|
|
assert exc_info.value is expected
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {"name": "RuntimeError", "message": "responder failed"},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("streamed", [False, True])
|
|
async def test_scripted_model_redacts_generation_span_errors_from_run_config(
|
|
streamed: bool,
|
|
) -> None:
|
|
expected = RuntimeError("sensitive provider payload")
|
|
model = ScriptedModel([ModelStep.raise_error(expected)], emit_traces=True)
|
|
agent = Agent(name="test", model=model)
|
|
run_config = RunConfig(trace_include_sensitive_data=False)
|
|
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
if streamed:
|
|
result = Runner.run_streamed(agent, "hi", run_config=run_config)
|
|
async for _event in result.stream_events():
|
|
pass
|
|
else:
|
|
await Runner.run(agent, "hi", run_config=run_config)
|
|
|
|
assert exc_info.value is expected
|
|
assert fetch_span_errors("generation") == [
|
|
{
|
|
"message": "Error",
|
|
"data": {
|
|
"name": "RuntimeError",
|
|
"message": REDACTED_TRACE_ERROR_MESSAGE,
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("call_id", [None, "", 123])
|
|
async def test_scripted_model_rejects_invalid_apply_patch_call_id(call_id: Any) -> None:
|
|
output = cast(
|
|
TResponseOutputItem,
|
|
{
|
|
"type": "apply_patch_call",
|
|
"id": "patch_item",
|
|
"call_id": call_id,
|
|
"operation": {"type": "delete_file", "path": "example.txt"},
|
|
},
|
|
)
|
|
model = ScriptedModel([[output]])
|
|
|
|
with pytest.raises(
|
|
ModelBehaviorError,
|
|
match="Tool invocations require a non-empty string call ID before execution",
|
|
):
|
|
await Runner.run(Agent(name="test", model=model), "hi")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("item_id", [None, "", 123])
|
|
async def test_scripted_model_rejects_invalid_apply_patch_item_id(item_id: Any) -> None:
|
|
output = cast(
|
|
TResponseOutputItem,
|
|
{
|
|
"type": "apply_patch_call",
|
|
"id": item_id,
|
|
"call_id": "call_1",
|
|
"operation": {"type": "delete_file", "path": "example.txt"},
|
|
},
|
|
)
|
|
model = ScriptedModel([[output]])
|
|
|
|
with pytest.raises(
|
|
ModelBehaviorError,
|
|
match="Apply-patch tool calls require a non-empty string item ID when provided",
|
|
):
|
|
await Runner.run(Agent(name="test", model=model), "hi")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_defaults_omitted_apply_patch_item_id_to_call_id() -> None:
|
|
output = cast(
|
|
TResponseOutputItem,
|
|
{
|
|
"type": "apply_patch_call",
|
|
"call_id": "call_1",
|
|
"operation": {"type": "delete_file", "path": "example.txt"},
|
|
},
|
|
)
|
|
model = ScriptedModel([[output]])
|
|
|
|
response = await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
item = response.output[0]
|
|
assert isinstance(item, ResponseApplyPatchToolCall)
|
|
assert item.id == "call_1"
|
|
assert item.status == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("status", [None, "", "invalid"])
|
|
async def test_scripted_model_rejects_invalid_explicit_apply_patch_status(status: Any) -> None:
|
|
output = cast(
|
|
TResponseOutputItem,
|
|
{
|
|
"type": "apply_patch_call",
|
|
"id": "patch_item",
|
|
"call_id": "call_1",
|
|
"status": status,
|
|
"operation": {"type": "delete_file", "path": "example.txt"},
|
|
},
|
|
)
|
|
model = ScriptedModel([[output]])
|
|
|
|
with pytest.raises(ValidationError):
|
|
await model.get_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_function_argument_events_use_output_item_id() -> None:
|
|
model = ScriptedModel([[function_call("lookup", "{}", call_id="call_1", item_id="item_1")]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
argument_events = [
|
|
event
|
|
for event in events
|
|
if isinstance(
|
|
event,
|
|
ResponseFunctionCallArgumentsDeltaEvent | ResponseFunctionCallArgumentsDoneEvent,
|
|
)
|
|
]
|
|
assert [event.item_id for event in argument_events] == ["item_1", "item_1"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_function_argument_events_preserve_empty_output_item_id() -> None:
|
|
output = function_call("lookup", "{}", call_id="call_1", item_id="")
|
|
model = ScriptedModel([[output]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
item_events = [
|
|
event
|
|
for event in events
|
|
if isinstance(event, ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent)
|
|
]
|
|
argument_events = [
|
|
event
|
|
for event in events
|
|
if isinstance(
|
|
event,
|
|
ResponseFunctionCallArgumentsDeltaEvent | ResponseFunctionCallArgumentsDoneEvent,
|
|
)
|
|
]
|
|
assert [event.item.id for event in item_events] == ["", ""]
|
|
assert [event.item_id for event in argument_events] == ["", ""]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_function_argument_events_fall_back_to_call_id() -> None:
|
|
output = ResponseFunctionToolCall(
|
|
type="function_call",
|
|
name="lookup",
|
|
arguments="{}",
|
|
call_id="call_1",
|
|
)
|
|
model = ScriptedModel([[output]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
argument_events = [
|
|
event
|
|
for event in events
|
|
if isinstance(
|
|
event,
|
|
ResponseFunctionCallArgumentsDeltaEvent | ResponseFunctionCallArgumentsDoneEvent,
|
|
)
|
|
]
|
|
assert [event.item_id for event in argument_events] == ["call_1", "call_1"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_uses_in_progress_added_payloads() -> None:
|
|
reasoning = ResponseReasoningItem(
|
|
id="reasoning_1",
|
|
summary=[Summary(text="summary", type="summary_text")],
|
|
encrypted_content="encrypted-reasoning",
|
|
type="reasoning",
|
|
status="completed",
|
|
)
|
|
message = assistant_message("hello", item_id="message_1")
|
|
function = function_call(
|
|
"lookup",
|
|
{"city": "Tokyo"},
|
|
call_id="call_1",
|
|
item_id="function_1",
|
|
)
|
|
completed_items = [reasoning, message, function]
|
|
usage = Usage(requests=3, input_tokens=4, output_tokens=5, total_tokens=9)
|
|
model = ScriptedModel([ModelStep(output=completed_items, usage=usage)])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
opening_responses = [
|
|
event.response
|
|
for event in events
|
|
if isinstance(event, ResponseCreatedEvent | ResponseInProgressEvent)
|
|
]
|
|
assert [response.output for response in opening_responses] == [[], []]
|
|
assert [response.status for response in opening_responses] == ["in_progress", "in_progress"]
|
|
assert [response.usage for response in opening_responses] == [None, None]
|
|
|
|
added_items = [
|
|
event.item for event in events if isinstance(event, ResponseOutputItemAddedEvent)
|
|
]
|
|
assert isinstance(added_items[0], ResponseReasoningItem)
|
|
assert added_items[0].encrypted_content is None
|
|
assert added_items[0].summary == []
|
|
assert added_items[0].status == "in_progress"
|
|
assert isinstance(added_items[1], ResponseOutputMessage)
|
|
assert added_items[1].content == []
|
|
assert added_items[1].status == "in_progress"
|
|
assert isinstance(added_items[2], ResponseFunctionToolCall)
|
|
assert added_items[2].arguments == ""
|
|
assert added_items[2].status == "in_progress"
|
|
|
|
added_summary_parts = [
|
|
event.part for event in events if isinstance(event, ResponseReasoningSummaryPartAddedEvent)
|
|
]
|
|
assert [part.text for part in added_summary_parts] == [""]
|
|
added_content_parts = [
|
|
event.part for event in events if isinstance(event, ResponseContentPartAddedEvent)
|
|
]
|
|
assert [part.text for part in added_content_parts] == [""]
|
|
|
|
done_items = [event.item for event in events if isinstance(event, ResponseOutputItemDoneEvent)]
|
|
assert done_items == completed_items
|
|
assert cast(ResponseReasoningItem, done_items[0]).encrypted_content == "encrypted-reasoning"
|
|
completed_event = next(event for event in events if isinstance(event, ResponseCompletedEvent))
|
|
assert completed_event.response.status == "completed"
|
|
assert completed_event.response.output == completed_items
|
|
assert (
|
|
cast(ResponseReasoningItem, completed_event.response.output[0]).encrypted_content
|
|
== "encrypted-reasoning"
|
|
)
|
|
assert completed_event.response.usage is not None
|
|
assert completed_event.response.usage.input_tokens == 4
|
|
assert completed_event.response.usage.output_tokens == 5
|
|
assert completed_event.response.usage.total_tokens == 9
|
|
request_count_attribute = "_agents_sdk_request_count"
|
|
assert getattr(completed_event.response.usage, request_count_attribute) == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_detaches_done_item_from_terminal_response() -> None:
|
|
model = ScriptedModel([[assistant_message("original", item_id="message_1")]])
|
|
completed_event: ResponseCompletedEvent | None = None
|
|
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
if isinstance(event, ResponseOutputItemDoneEvent):
|
|
assert isinstance(event.item, ResponseOutputMessage)
|
|
assert isinstance(event.item.content[0], ResponseOutputText)
|
|
event.item.content[0].text = "mutated"
|
|
elif isinstance(event, ResponseCompletedEvent):
|
|
completed_event = event
|
|
|
|
assert completed_event is not None
|
|
terminal_item = completed_event.response.output[0]
|
|
assert isinstance(terminal_item, ResponseOutputMessage)
|
|
assert isinstance(terminal_item.content[0], ResponseOutputText)
|
|
assert terminal_item.content[0].text == "original"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_detaches_content_part_event_payloads() -> None:
|
|
text = ResponseOutputText.model_validate(
|
|
{
|
|
"type": "output_text",
|
|
"text": "original text",
|
|
"annotations": [],
|
|
"logprobs": [],
|
|
"provider_data": {"nested": ["original text"]},
|
|
}
|
|
)
|
|
refusal = ResponseOutputRefusal.model_validate(
|
|
{
|
|
"type": "refusal",
|
|
"refusal": "original refusal",
|
|
"provider_data": {"nested": ["original refusal"]},
|
|
}
|
|
)
|
|
message = ResponseOutputMessage(
|
|
id="message_1",
|
|
type="message",
|
|
role="assistant",
|
|
status="completed",
|
|
content=[text, refusal],
|
|
)
|
|
model = ScriptedModel([[message]])
|
|
done_parts: list[ResponseOutputText | ResponseOutputRefusal] = []
|
|
done_item: ResponseOutputMessage | None = None
|
|
completed_item: ResponseOutputMessage | None = None
|
|
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
if isinstance(event, ResponseContentPartAddedEvent):
|
|
cast(Any, event.part).provider_data["nested"][0] = "mutated"
|
|
elif isinstance(event, ResponseContentPartDoneEvent):
|
|
done_parts.append(event.part)
|
|
elif isinstance(event, ResponseOutputItemDoneEvent):
|
|
assert isinstance(event.item, ResponseOutputMessage)
|
|
done_item = event.item
|
|
elif isinstance(event, ResponseCompletedEvent):
|
|
assert isinstance(event.response.output[0], ResponseOutputMessage)
|
|
completed_item = event.response.output[0]
|
|
|
|
expected_provider_data = [
|
|
{"nested": ["original text"]},
|
|
{"nested": ["original refusal"]},
|
|
]
|
|
assert [cast(Any, part).provider_data for part in done_parts] == expected_provider_data
|
|
assert done_item is not None
|
|
assert [cast(Any, part).provider_data for part in done_item.content] == expected_provider_data
|
|
assert completed_item is not None
|
|
assert [
|
|
cast(Any, part).provider_data for part in completed_item.content
|
|
] == expected_provider_data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_preserves_text_logprobs() -> None:
|
|
output_logprobs = [
|
|
Logprob(
|
|
token="hello",
|
|
bytes=[104, 101, 108, 108, 111],
|
|
logprob=-0.25,
|
|
top_logprobs=[
|
|
LogprobTopLogprob(
|
|
token="hi",
|
|
bytes=[104, 105],
|
|
logprob=-1.5,
|
|
)
|
|
],
|
|
),
|
|
Logprob(
|
|
token="!",
|
|
bytes=[33],
|
|
logprob=-0.1,
|
|
top_logprobs=[],
|
|
),
|
|
]
|
|
text = ResponseOutputText(
|
|
type="output_text",
|
|
text="hello",
|
|
annotations=[],
|
|
logprobs=output_logprobs,
|
|
)
|
|
message = ResponseOutputMessage(
|
|
id="message_1",
|
|
type="message",
|
|
role="assistant",
|
|
status="completed",
|
|
content=[text],
|
|
)
|
|
model = ScriptedModel([[message]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
added_event = next(
|
|
event for event in events if isinstance(event, ResponseContentPartAddedEvent)
|
|
)
|
|
delta_event = next(event for event in events if isinstance(event, ResponseTextDeltaEvent))
|
|
done_event = next(event for event in events if isinstance(event, ResponseTextDoneEvent))
|
|
assert isinstance(added_event.part, ResponseOutputText)
|
|
assert added_event.part.logprobs == []
|
|
assert [(logprob.token, logprob.logprob) for logprob in delta_event.logprobs] == [
|
|
("hello", -0.25),
|
|
("!", -0.1),
|
|
]
|
|
assert delta_event.logprobs[0].top_logprobs is not None
|
|
assert [logprob.token for logprob in delta_event.logprobs[0].top_logprobs] == ["hi"]
|
|
assert delta_event.logprobs[1].top_logprobs == []
|
|
assert [(logprob.token, logprob.logprob) for logprob in done_event.logprobs] == [
|
|
("hello", -0.25),
|
|
("!", -0.1),
|
|
]
|
|
assert done_event.logprobs[0].top_logprobs is not None
|
|
assert [logprob.token for logprob in done_event.logprobs[0].top_logprobs] == ["hi"]
|
|
assert done_event.logprobs[1].top_logprobs == []
|
|
|
|
completed_event = next(event for event in events if isinstance(event, ResponseCompletedEvent))
|
|
completed_message = completed_event.response.output[0]
|
|
assert isinstance(completed_message, ResponseOutputMessage)
|
|
completed_text = completed_message.content[0]
|
|
assert isinstance(completed_text, ResponseOutputText)
|
|
assert completed_text.logprobs == output_logprobs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_emits_text_annotation_events() -> None:
|
|
annotations = [
|
|
AnnotationURLCitation(
|
|
end_index=5,
|
|
start_index=0,
|
|
title="Example",
|
|
type="url_citation",
|
|
url="https://example.test",
|
|
),
|
|
AnnotationFilePath(
|
|
file_id="file_1",
|
|
index=6,
|
|
type="file_path",
|
|
),
|
|
]
|
|
text = ResponseOutputText(
|
|
type="output_text",
|
|
text="hello file",
|
|
annotations=annotations,
|
|
logprobs=[],
|
|
)
|
|
message = ResponseOutputMessage(
|
|
id="message_1",
|
|
type="message",
|
|
role="assistant",
|
|
status="completed",
|
|
content=[text],
|
|
)
|
|
model = ScriptedModel([[message]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
added_part = next(event for event in events if isinstance(event, ResponseContentPartAddedEvent))
|
|
annotation_events = [
|
|
event for event in events if isinstance(event, ResponseOutputTextAnnotationAddedEvent)
|
|
]
|
|
text_done = next(event for event in events if isinstance(event, ResponseTextDoneEvent))
|
|
assert isinstance(added_part.part, ResponseOutputText)
|
|
assert added_part.part.annotations == []
|
|
assert [event.annotation_index for event in annotation_events] == [0, 1]
|
|
assert [event.model_dump()["annotation"] for event in annotation_events] == [
|
|
annotation.model_dump() for annotation in annotations
|
|
]
|
|
assert all(event.item_id == "message_1" for event in annotation_events)
|
|
assert all(event.output_index == 0 for event in annotation_events)
|
|
assert all(event.content_index == 0 for event in annotation_events)
|
|
assert annotation_events[0].sequence_number == added_part.sequence_number + 2
|
|
assert annotation_events[1].sequence_number == annotation_events[0].sequence_number + 1
|
|
assert text_done.sequence_number == annotation_events[1].sequence_number + 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_emits_reasoning_content_events() -> None:
|
|
reasoning = ResponseReasoningItem(
|
|
id="reasoning_1",
|
|
summary=[],
|
|
content=[Content(text="think carefully", type="reasoning_text")],
|
|
type="reasoning",
|
|
status="completed",
|
|
)
|
|
model = ScriptedModel([[reasoning]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
added_event = next(event for event in events if isinstance(event, ResponseOutputItemAddedEvent))
|
|
assert isinstance(added_event.item, ResponseReasoningItem)
|
|
assert added_event.item.content == []
|
|
delta_event = next(
|
|
event for event in events if isinstance(event, ResponseReasoningTextDeltaEvent)
|
|
)
|
|
done_event = next(
|
|
event for event in events if isinstance(event, ResponseReasoningTextDoneEvent)
|
|
)
|
|
assert delta_event.item_id == "reasoning_1"
|
|
assert delta_event.output_index == 0
|
|
assert delta_event.content_index == 0
|
|
assert delta_event.delta == "think carefully"
|
|
assert done_event.item_id == "reasoning_1"
|
|
assert done_event.output_index == 0
|
|
assert done_event.content_index == 0
|
|
assert done_event.text == "think carefully"
|
|
assert done_event.sequence_number == delta_event.sequence_number + 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_marks_added_apply_patch_call_in_progress() -> None:
|
|
apply_patch_call = ResponseApplyPatchToolCall(
|
|
type="apply_patch_call",
|
|
id="apply_patch_1",
|
|
call_id="call_1",
|
|
status="completed",
|
|
operation=cast(
|
|
Any,
|
|
{"type": "update_file", "path": "test.md", "diff": "-old\n+new\n"},
|
|
),
|
|
)
|
|
model = ScriptedModel([[apply_patch_call]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
added_event = next(event for event in events if isinstance(event, ResponseOutputItemAddedEvent))
|
|
done_event = next(event for event in events if isinstance(event, ResponseOutputItemDoneEvent))
|
|
completed_event = next(event for event in events if isinstance(event, ResponseCompletedEvent))
|
|
|
|
assert isinstance(added_event.item, ResponseApplyPatchToolCall)
|
|
assert added_event.item.status == "in_progress"
|
|
assert isinstance(done_event.item, ResponseApplyPatchToolCall)
|
|
assert done_event.item.status == "completed"
|
|
assert isinstance(completed_event.response.output[0], ResponseApplyPatchToolCall)
|
|
assert completed_event.response.output[0].status == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_automatic_stream_emits_refusal_content_events() -> None:
|
|
refusal = ResponseOutputRefusal(type="refusal", refusal="I cannot help with that.")
|
|
message = ResponseOutputMessage(
|
|
id="message_1",
|
|
type="message",
|
|
role="assistant",
|
|
status="completed",
|
|
content=[refusal],
|
|
)
|
|
model = ScriptedModel([[message]])
|
|
|
|
events = [
|
|
event
|
|
async for event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
)
|
|
]
|
|
|
|
assert [event.type for event in events] == [
|
|
"response.created",
|
|
"response.in_progress",
|
|
"response.output_item.added",
|
|
"response.content_part.added",
|
|
"response.refusal.delta",
|
|
"response.refusal.done",
|
|
"response.content_part.done",
|
|
"response.output_item.done",
|
|
"response.completed",
|
|
]
|
|
assert [event.sequence_number for event in events] == list(range(len(events)))
|
|
content_events = [
|
|
event
|
|
for event in events
|
|
if isinstance(
|
|
event,
|
|
ResponseContentPartAddedEvent
|
|
| ResponseRefusalDeltaEvent
|
|
| ResponseRefusalDoneEvent
|
|
| ResponseContentPartDoneEvent,
|
|
)
|
|
]
|
|
assert [event.type for event in content_events] == [
|
|
"response.content_part.added",
|
|
"response.refusal.delta",
|
|
"response.refusal.done",
|
|
"response.content_part.done",
|
|
]
|
|
assert all(event.item_id == "message_1" for event in content_events)
|
|
assert all(event.output_index == 0 for event in content_events)
|
|
assert all(event.content_index == 0 for event in content_events)
|
|
added, delta, refusal_done, content_done = content_events
|
|
assert isinstance(added, ResponseContentPartAddedEvent)
|
|
assert isinstance(added.part, ResponseOutputRefusal)
|
|
assert added.part.refusal == ""
|
|
assert isinstance(delta, ResponseRefusalDeltaEvent)
|
|
assert delta.delta == "I cannot help with that."
|
|
assert isinstance(refusal_done, ResponseRefusalDoneEvent)
|
|
assert refusal_done.refusal == "I cannot help with that."
|
|
assert isinstance(content_done, ResponseContentPartDoneEvent)
|
|
assert content_done.part == refusal
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_rejects_unexpected_call() -> None:
|
|
model = ScriptedModel()
|
|
|
|
with pytest.raises(UnexpectedModelCall, match="no scripted steps remain") as exc_info:
|
|
await Runner.run(Agent(name="test", model=model), "hi")
|
|
|
|
assert exc_info.value.call_index == 0
|
|
assert "call #1" in str(exc_info.value)
|
|
assert exc_info.value.call.streamed is False
|
|
assert exc_info.value.call.input == [{"content": "hi", "role": "user"}]
|
|
|
|
exc_info.value.call.input[0]["content"] = "changed"
|
|
assert model.calls[0].input == [{"content": "hi", "role": "user"}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scripted_model_unexpected_streaming_call_records_streamed_attribute() -> None:
|
|
model = ScriptedModel()
|
|
|
|
with pytest.raises(UnexpectedModelCall) as exc_info:
|
|
async for _event in model.stream_response(
|
|
None,
|
|
[],
|
|
ModelSettings(),
|
|
[],
|
|
None,
|
|
[],
|
|
ModelTracing.DISABLED,
|
|
previous_response_id=None,
|
|
conversation_id=None,
|
|
prompt=None,
|
|
):
|
|
pass
|
|
|
|
assert exc_info.value.call_index == 0
|
|
assert exc_info.value.call.streamed is True
|
|
|
|
|
|
def test_scripted_model_reports_unconsumed_steps() -> None:
|
|
model = ScriptedModel([[assistant_message("unused")]])
|
|
|
|
with pytest.raises(UnconsumedModelSteps, match="1 scripted model step") as exc_info:
|
|
model.assert_complete()
|
|
|
|
assert exc_info.value.remaining_steps == 1
|