wehub-resource-sync
727 lines
26 KiB
Python
727 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import gc
|
|
import json
|
|
import weakref
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
from openai.types.responses.computer_action import Click as BatchedClick, Type as BatchedType
|
|
from openai.types.responses.response_computer_tool_call import (
|
|
ActionScreenshot,
|
|
ResponseComputerToolCall,
|
|
)
|
|
from openai.types.responses.response_computer_tool_call_param import ResponseComputerToolCallParam
|
|
from openai.types.responses.response_file_search_tool_call import ResponseFileSearchToolCall
|
|
from openai.types.responses.response_file_search_tool_call_param import (
|
|
ResponseFileSearchToolCallParam,
|
|
)
|
|
from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
|
|
from openai.types.responses.response_function_tool_call_param import ResponseFunctionToolCallParam
|
|
from openai.types.responses.response_function_web_search import (
|
|
ActionSearch,
|
|
ResponseFunctionWebSearch,
|
|
)
|
|
from openai.types.responses.response_function_web_search_param import ResponseFunctionWebSearchParam
|
|
from openai.types.responses.response_input_item_param import ResponseInputItemParam
|
|
from openai.types.responses.response_output_message import ResponseOutputMessage
|
|
from openai.types.responses.response_output_message_param import ResponseOutputMessageParam
|
|
from openai.types.responses.response_output_refusal import ResponseOutputRefusal
|
|
from openai.types.responses.response_output_text import ResponseOutputText
|
|
from openai.types.responses.response_output_text_param import ResponseOutputTextParam
|
|
from openai.types.responses.response_reasoning_item import ResponseReasoningItem, Summary
|
|
from openai.types.responses.response_reasoning_item_param import ResponseReasoningItemParam
|
|
from openai.types.responses.response_tool_search_call import ResponseToolSearchCall
|
|
from openai.types.responses.response_tool_search_output_item import ResponseToolSearchOutputItem
|
|
from pydantic import TypeAdapter, ValidationError
|
|
|
|
from agents import (
|
|
Agent,
|
|
HandoffOutputItem,
|
|
ItemHelpers,
|
|
MessageOutputItem,
|
|
ModelResponse,
|
|
ReasoningItem,
|
|
RunItem,
|
|
TResponseInputItem,
|
|
Usage,
|
|
)
|
|
from agents.items import ToolCallItem, ToolCallOutputItem
|
|
|
|
|
|
def make_message(
|
|
content_items: list[ResponseOutputText | ResponseOutputRefusal],
|
|
) -> ResponseOutputMessage:
|
|
"""
|
|
Helper to construct a ResponseOutputMessage with a single batch of content
|
|
items, using a fixed id/status.
|
|
"""
|
|
return ResponseOutputMessage(
|
|
id="msg123",
|
|
content=content_items,
|
|
role="assistant",
|
|
status="completed",
|
|
type="message",
|
|
)
|
|
|
|
|
|
def test_extract_last_content_of_text_message() -> None:
|
|
# Build a message containing two text segments.
|
|
content1 = ResponseOutputText(annotations=[], text="Hello ", type="output_text", logprobs=[])
|
|
content2 = ResponseOutputText(annotations=[], text="world!", type="output_text", logprobs=[])
|
|
message = make_message([content1, content2])
|
|
# Helpers should yield the last segment's text.
|
|
assert ItemHelpers.extract_last_content(message) == "world!"
|
|
|
|
|
|
def test_extract_last_content_of_refusal_message() -> None:
|
|
# Build a message whose last content entry is a refusal.
|
|
content1 = ResponseOutputText(
|
|
annotations=[], text="Before refusal", type="output_text", logprobs=[]
|
|
)
|
|
refusal = ResponseOutputRefusal(refusal="I cannot do that", type="refusal")
|
|
message = make_message([content1, refusal])
|
|
# Helpers should extract the refusal string when last content is a refusal.
|
|
assert ItemHelpers.extract_last_content(message) == "I cannot do that"
|
|
|
|
|
|
def test_none_refusal_is_rejected_before_extract_last_content() -> None:
|
|
with pytest.raises(ValidationError, match="refusal"):
|
|
ResponseOutputRefusal.model_validate({"refusal": None, "type": "refusal"})
|
|
|
|
|
|
def test_extract_last_content_non_message_returns_empty() -> None:
|
|
# Construct some other type of output item, e.g. a tool call, to verify non-message returns "".
|
|
tool_call = ResponseFunctionToolCall(
|
|
id="tool123",
|
|
arguments="{}",
|
|
call_id="call123",
|
|
name="func",
|
|
type="function_call",
|
|
)
|
|
assert ItemHelpers.extract_last_content(tool_call) == ""
|
|
|
|
|
|
def test_extract_last_text_returns_text_only() -> None:
|
|
# A message whose last segment is text yields the text.
|
|
first_text = ResponseOutputText(annotations=[], text="part1", type="output_text", logprobs=[])
|
|
second_text = ResponseOutputText(annotations=[], text="part2", type="output_text", logprobs=[])
|
|
message = make_message([first_text, second_text])
|
|
assert ItemHelpers.extract_last_text(message) == "part2"
|
|
# Whereas when last content is a refusal, extract_last_text returns None.
|
|
message2 = make_message([first_text, ResponseOutputRefusal(refusal="no", type="refusal")])
|
|
assert ItemHelpers.extract_last_text(message2) is None
|
|
|
|
|
|
def test_extract_text_concatenates_all_text_segments() -> None:
|
|
first_text = ResponseOutputText(annotations=[], text="part1", type="output_text", logprobs=[])
|
|
second_text = ResponseOutputText(annotations=[], text="part2", type="output_text", logprobs=[])
|
|
refusal = ResponseOutputRefusal(refusal="no", type="refusal")
|
|
message = make_message([first_text, refusal, second_text])
|
|
|
|
assert ItemHelpers.extract_text(message) == "part1part2"
|
|
assert (
|
|
ItemHelpers.extract_text(
|
|
ResponseFunctionToolCall(
|
|
id="tool123",
|
|
arguments="{}",
|
|
call_id="call123",
|
|
name="func",
|
|
type="function_call",
|
|
)
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_extract_text_tolerates_none_text_content() -> None:
|
|
"""Regression: ``content_item.text`` can be ``None`` when output items
|
|
are assembled via ``model_construct`` (e.g. partial streaming responses)
|
|
or surfaced through provider gateways like LiteLLM. Without the ``or ""``
|
|
guard, ``extract_text`` raised
|
|
``TypeError: can only concatenate str (not "NoneType") to str`` deep
|
|
inside ``execute_tools_and_side_effects`` and aborted the agent turn.
|
|
"""
|
|
none_text = ResponseOutputText.model_construct(
|
|
annotations=[], text=None, type="output_text", logprobs=[]
|
|
)
|
|
real_text = ResponseOutputText(annotations=[], text="hello", type="output_text", logprobs=[])
|
|
|
|
# Single None-text item: result is None (since concatenated text is "").
|
|
assert ItemHelpers.extract_text(make_message([none_text])) is None
|
|
|
|
# Mixed content: real text is preserved, None is skipped.
|
|
assert ItemHelpers.extract_text(make_message([real_text, none_text])) == "hello"
|
|
assert ItemHelpers.extract_text(make_message([none_text, real_text])) == "hello"
|
|
|
|
|
|
def test_input_to_new_input_list_from_string() -> None:
|
|
result = ItemHelpers.input_to_new_input_list("hi")
|
|
# Should wrap the string into a list with a single dict containing content and user role.
|
|
assert isinstance(result, list)
|
|
assert result == [{"content": "hi", "role": "user"}]
|
|
|
|
|
|
def test_input_to_new_input_list_deep_copies_lists() -> None:
|
|
# Given a list of message dictionaries, ensure the returned list is a deep copy.
|
|
original: list[TResponseInputItem] = [{"content": "abc", "role": "developer"}]
|
|
new_list = ItemHelpers.input_to_new_input_list(original)
|
|
assert new_list == original
|
|
# Mutating the returned list should not mutate the original.
|
|
new_list.pop()
|
|
assert "content" in original[0] and original[0].get("content") == "abc"
|
|
|
|
|
|
def test_text_message_output_concatenates_text_segments() -> None:
|
|
# Build a message with both text and refusal segments, only text segments are concatenated.
|
|
pieces: list[ResponseOutputText | ResponseOutputRefusal] = []
|
|
pieces.append(ResponseOutputText(annotations=[], text="a", type="output_text", logprobs=[]))
|
|
pieces.append(ResponseOutputRefusal(refusal="denied", type="refusal"))
|
|
pieces.append(ResponseOutputText(annotations=[], text="b", type="output_text", logprobs=[]))
|
|
message = make_message(pieces)
|
|
# Wrap into MessageOutputItem to feed into text_message_output.
|
|
item = MessageOutputItem(agent=Agent(name="test"), raw_item=message)
|
|
assert ItemHelpers.text_message_output(item) == "ab"
|
|
|
|
|
|
def test_text_message_outputs_across_list_of_runitems() -> None:
|
|
"""
|
|
Compose several RunItem instances, including a non-message run item, and ensure
|
|
that only MessageOutputItem instances contribute any text. The non-message
|
|
(ReasoningItem) should be ignored by Helpers.text_message_outputs.
|
|
"""
|
|
message1 = make_message(
|
|
[ResponseOutputText(annotations=[], text="foo", type="output_text", logprobs=[])]
|
|
)
|
|
message2 = make_message(
|
|
[ResponseOutputText(annotations=[], text="bar", type="output_text", logprobs=[])]
|
|
)
|
|
item1: RunItem = MessageOutputItem(agent=Agent(name="test"), raw_item=message1)
|
|
item2: RunItem = MessageOutputItem(agent=Agent(name="test"), raw_item=message2)
|
|
# Create a non-message run item of a different type, e.g., a reasoning trace.
|
|
reasoning = ResponseReasoningItem(id="rid", summary=[], type="reasoning")
|
|
non_message_item: RunItem = ReasoningItem(agent=Agent(name="test"), raw_item=reasoning)
|
|
# Confirm only the message outputs are concatenated.
|
|
assert ItemHelpers.text_message_outputs([item1, non_message_item, item2]) == "foobar"
|
|
|
|
|
|
def test_message_output_item_retains_agent_until_release() -> None:
|
|
# Construct the run item with an inline agent to ensure the run item keeps a strong reference.
|
|
message = make_message([ResponseOutputText(annotations=[], text="hello", type="output_text")])
|
|
agent = Agent(name="inline")
|
|
item = MessageOutputItem(agent=agent, raw_item=message)
|
|
assert item.agent is agent
|
|
assert item.agent.name == "inline"
|
|
|
|
# Releasing the agent should keep the weak reference alive while strong refs remain.
|
|
item.release_agent()
|
|
assert item.agent is agent
|
|
|
|
agent_ref = weakref.ref(agent)
|
|
del agent
|
|
gc.collect()
|
|
|
|
# Once the original agent is collected, the weak reference should drop.
|
|
assert agent_ref() is None
|
|
assert item.agent is None
|
|
|
|
|
|
def test_handoff_output_item_retains_agents_until_gc() -> None:
|
|
raw_item: TResponseInputItem = {
|
|
"call_id": "call1",
|
|
"output": "handoff",
|
|
"type": "function_call_output",
|
|
}
|
|
owner_agent = Agent(name="owner")
|
|
source_agent = Agent(name="source")
|
|
target_agent = Agent(name="target")
|
|
item = HandoffOutputItem(
|
|
agent=owner_agent,
|
|
raw_item=raw_item,
|
|
source_agent=source_agent,
|
|
target_agent=target_agent,
|
|
)
|
|
|
|
item.release_agent()
|
|
assert item.agent is owner_agent
|
|
assert item.source_agent is source_agent
|
|
assert item.target_agent is target_agent
|
|
|
|
owner_ref = weakref.ref(owner_agent)
|
|
source_ref = weakref.ref(source_agent)
|
|
target_ref = weakref.ref(target_agent)
|
|
del owner_agent
|
|
del source_agent
|
|
del target_agent
|
|
gc.collect()
|
|
|
|
assert owner_ref() is None
|
|
assert source_ref() is None
|
|
assert target_ref() is None
|
|
assert item.agent is None
|
|
assert item.source_agent is None
|
|
assert item.target_agent is None
|
|
|
|
|
|
def test_handoff_output_item_converts_protocol_payload() -> None:
|
|
raw_item = cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call-123",
|
|
"output": "ok",
|
|
},
|
|
)
|
|
owner_agent = Agent(name="owner")
|
|
source_agent = Agent(name="source")
|
|
target_agent = Agent(name="target")
|
|
item = HandoffOutputItem(
|
|
agent=owner_agent,
|
|
raw_item=raw_item,
|
|
source_agent=source_agent,
|
|
target_agent=target_agent,
|
|
)
|
|
|
|
converted = item.to_input_item()
|
|
assert converted["type"] == "function_call_output"
|
|
assert converted["call_id"] == "call-123"
|
|
assert converted["output"] == "ok"
|
|
|
|
|
|
def test_handoff_output_item_stringifies_object_output() -> None:
|
|
raw_item = cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": "call-obj",
|
|
"output": {"assistant": "Weather Assistant"},
|
|
},
|
|
)
|
|
owner_agent = Agent(name="owner")
|
|
source_agent = Agent(name="source")
|
|
target_agent = Agent(name="target")
|
|
item = HandoffOutputItem(
|
|
agent=owner_agent,
|
|
raw_item=raw_item,
|
|
source_agent=source_agent,
|
|
target_agent=target_agent,
|
|
)
|
|
|
|
converted = item.to_input_item()
|
|
assert converted["type"] == "function_call_output"
|
|
assert converted["call_id"] == "call-obj"
|
|
assert isinstance(converted["output"], dict)
|
|
assert converted["output"] == {"assistant": "Weather Assistant"}
|
|
|
|
|
|
def test_tool_call_output_item_preserves_function_output_structure() -> None:
|
|
agent = Agent(name="tester")
|
|
raw_item = {
|
|
"type": "function_call_output",
|
|
"call_id": "call-keep",
|
|
"output": [{"type": "output_text", "text": "value"}],
|
|
}
|
|
item = ToolCallOutputItem(agent=agent, raw_item=raw_item, output="value")
|
|
|
|
payload = item.to_input_item()
|
|
assert isinstance(payload, dict)
|
|
assert payload["type"] == "function_call_output"
|
|
assert payload["output"] == raw_item["output"]
|
|
|
|
|
|
def test_tool_call_output_item_constructs_function_call_output_dict():
|
|
# Build a simple ResponseFunctionToolCall.
|
|
call = ResponseFunctionToolCall(
|
|
id="call-abc",
|
|
arguments='{"x": 1}',
|
|
call_id="call-abc",
|
|
name="do_something",
|
|
type="function_call",
|
|
)
|
|
payload = ItemHelpers.tool_call_output_item(call, "result-string")
|
|
|
|
assert isinstance(payload, dict)
|
|
assert payload["type"] == "function_call_output"
|
|
assert payload["call_id"] == call.id
|
|
assert payload["output"] == "result-string"
|
|
|
|
|
|
# The following tests ensure that every possible output item type defined by
|
|
# OpenAI's API can be converted back into an input item dict via
|
|
# ModelResponse.to_input_items. The output and input schema for each item are
|
|
# intended to be symmetric, so given any ResponseOutputItem, its model_dump
|
|
# should produce a dict that can satisfy the corresponding TypedDict input
|
|
# type. These tests construct minimal valid instances of each output type,
|
|
# invoke to_input_items, and then verify that the resulting dict can be used
|
|
# to round-trip back into a Pydantic output model without errors.
|
|
|
|
|
|
def test_to_input_items_for_message() -> None:
|
|
"""An output message should convert into an input dict matching the message's own structure."""
|
|
content = ResponseOutputText(
|
|
annotations=[], text="hello world", type="output_text", logprobs=[]
|
|
)
|
|
message = ResponseOutputMessage(
|
|
id="m1", content=[content], role="assistant", status="completed", type="message"
|
|
)
|
|
resp = ModelResponse(output=[message], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
# The dict should contain exactly the primitive values of the message
|
|
expected: ResponseOutputMessageParam = {
|
|
"id": "m1",
|
|
"content": [
|
|
{
|
|
"annotations": [],
|
|
"logprobs": [],
|
|
"text": "hello world",
|
|
"type": "output_text",
|
|
}
|
|
],
|
|
"role": "assistant",
|
|
"status": "completed",
|
|
"type": "message",
|
|
}
|
|
assert input_items[0] == expected
|
|
|
|
|
|
def test_to_input_items_for_function_call() -> None:
|
|
"""A function tool call output should produce the same dict as a function tool call input."""
|
|
tool_call = ResponseFunctionToolCall(
|
|
id="f1", arguments="{}", call_id="c1", name="func", type="function_call"
|
|
)
|
|
resp = ModelResponse(output=[tool_call], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
expected: ResponseFunctionToolCallParam = {
|
|
"id": "f1",
|
|
"arguments": "{}",
|
|
"call_id": "c1",
|
|
"name": "func",
|
|
"type": "function_call",
|
|
}
|
|
assert input_items[0] == expected
|
|
|
|
|
|
def test_to_input_items_for_file_search_call() -> None:
|
|
"""A file search tool call output should produce the same dict as a file search input."""
|
|
fs_call = ResponseFileSearchToolCall(
|
|
id="fs1", queries=["query"], status="completed", type="file_search_call"
|
|
)
|
|
resp = ModelResponse(output=[fs_call], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
expected: ResponseFileSearchToolCallParam = {
|
|
"id": "fs1",
|
|
"queries": ["query"],
|
|
"status": "completed",
|
|
"type": "file_search_call",
|
|
}
|
|
assert input_items[0] == expected
|
|
|
|
|
|
def test_to_input_items_for_web_search_call() -> None:
|
|
"""A web search tool call output should produce the same dict as a web search input."""
|
|
ws_call = ResponseFunctionWebSearch(
|
|
id="w1",
|
|
action=ActionSearch(type="search", query="query"),
|
|
status="completed",
|
|
type="web_search_call",
|
|
)
|
|
resp = ModelResponse(output=[ws_call], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
expected: ResponseFunctionWebSearchParam = {
|
|
"id": "w1",
|
|
"status": "completed",
|
|
"type": "web_search_call",
|
|
"action": {"type": "search", "query": "query"},
|
|
}
|
|
assert input_items[0] == expected
|
|
|
|
|
|
def test_to_input_items_for_computer_call_click() -> None:
|
|
"""A computer call output should yield a dict whose shape matches the computer call input."""
|
|
action = ActionScreenshot(type="screenshot")
|
|
comp_call = ResponseComputerToolCall(
|
|
id="comp1",
|
|
action=action,
|
|
type="computer_call",
|
|
call_id="comp1",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
resp = ModelResponse(output=[comp_call], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
converted_dict = input_items[0]
|
|
# Top-level keys should match what we expect for a computer call input
|
|
expected: ResponseComputerToolCallParam = {
|
|
"id": "comp1",
|
|
"type": "computer_call",
|
|
"action": {"type": "screenshot"},
|
|
"call_id": "comp1",
|
|
"pending_safety_checks": [],
|
|
"status": "completed",
|
|
}
|
|
assert converted_dict == expected
|
|
|
|
|
|
def test_to_input_items_for_computer_call_batched_actions() -> None:
|
|
"""A batched computer call should preserve its actions list when replayed as input."""
|
|
comp_call = ResponseComputerToolCall(
|
|
id="comp2",
|
|
actions=[
|
|
BatchedClick(type="click", x=3, y=4, button="left"),
|
|
BatchedType(type="type", text="hello"),
|
|
],
|
|
type="computer_call",
|
|
call_id="comp2",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
resp = ModelResponse(output=[comp_call], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
assert input_items[0] == {
|
|
"id": "comp2",
|
|
"type": "computer_call",
|
|
"actions": [
|
|
{"type": "click", "x": 3, "y": 4, "button": "left"},
|
|
{"type": "type", "text": "hello"},
|
|
],
|
|
"call_id": "comp2",
|
|
"pending_safety_checks": [],
|
|
"status": "completed",
|
|
}
|
|
|
|
|
|
def test_to_input_items_for_reasoning() -> None:
|
|
"""A reasoning output should produce the same dict as a reasoning input item."""
|
|
rc = Summary(text="why", type="summary_text")
|
|
reasoning = ResponseReasoningItem(id="rid1", summary=[rc], type="reasoning")
|
|
resp = ModelResponse(output=[reasoning], usage=Usage(), response_id=None)
|
|
input_items = resp.to_input_items()
|
|
assert isinstance(input_items, list) and len(input_items) == 1
|
|
converted_dict = input_items[0]
|
|
|
|
expected: ResponseReasoningItemParam = {
|
|
"id": "rid1",
|
|
"summary": [{"text": "why", "type": "summary_text"}],
|
|
"type": "reasoning",
|
|
}
|
|
print(converted_dict)
|
|
print(expected)
|
|
assert converted_dict == expected
|
|
|
|
|
|
def test_to_input_items_for_tool_search_strips_created_by() -> None:
|
|
"""Tool-search output items should reuse the replay sanitizer before round-tripping."""
|
|
tool_search_call = ResponseToolSearchCall(
|
|
id="tsc_123",
|
|
call_id="call_tsc_123",
|
|
arguments={"query": "profile"},
|
|
execution="server",
|
|
status="completed",
|
|
type="tool_search_call",
|
|
created_by="server",
|
|
)
|
|
tool_search_output = ResponseToolSearchOutputItem(
|
|
id="tso_123",
|
|
call_id="call_tsc_123",
|
|
execution="server",
|
|
status="completed",
|
|
tools=[],
|
|
type="tool_search_output",
|
|
created_by="server",
|
|
)
|
|
|
|
resp = ModelResponse(
|
|
output=[tool_search_call, tool_search_output], usage=Usage(), response_id=None
|
|
)
|
|
input_items = resp.to_input_items()
|
|
|
|
assert input_items == [
|
|
{
|
|
"id": "tsc_123",
|
|
"call_id": "call_tsc_123",
|
|
"arguments": {"query": "profile"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"type": "tool_search_call",
|
|
},
|
|
{
|
|
"id": "tso_123",
|
|
"call_id": "call_tsc_123",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
"type": "tool_search_output",
|
|
},
|
|
]
|
|
|
|
|
|
def test_input_to_new_input_list_copies_the_ones_produced_by_pydantic() -> None:
|
|
"""Validated input items should be copied and made JSON dump compatible."""
|
|
original = ResponseOutputMessageParam(
|
|
id="a75654dc-7492-4d1c-bce0-89e8312fbdd7",
|
|
content=[
|
|
ResponseOutputTextParam(
|
|
type="output_text",
|
|
text="Hey, what's up?",
|
|
annotations=[],
|
|
logprobs=[],
|
|
)
|
|
],
|
|
role="assistant",
|
|
status="completed",
|
|
type="message",
|
|
)
|
|
validated = TypeAdapter(list[ResponseInputItemParam]).validate_python([original])
|
|
|
|
new_list = ItemHelpers.input_to_new_input_list(validated)
|
|
assert len(new_list) == 1
|
|
assert new_list[0]["id"] == original["id"] # type: ignore
|
|
assert new_list[0]["role"] == original["role"] # type: ignore
|
|
assert new_list[0]["status"] == original["status"] # type: ignore
|
|
assert new_list[0]["type"] == original["type"]
|
|
assert isinstance(new_list[0]["content"], list)
|
|
|
|
first_content = cast(dict[str, object], new_list[0]["content"][0])
|
|
assert first_content["type"] == "output_text"
|
|
assert first_content["text"] == "Hey, what's up?"
|
|
assert isinstance(first_content["annotations"], list)
|
|
assert isinstance(first_content["logprobs"], list)
|
|
|
|
# This used to fail when validated payloads retained ValidatorIterator fields.
|
|
json.dumps(new_list)
|
|
|
|
|
|
def test_tool_call_item_to_input_item_keeps_payload_api_safe() -> None:
|
|
agent = Agent(name="test", instructions="test")
|
|
raw_item = ResponseFunctionToolCall(
|
|
id="fc_1",
|
|
call_id="call_1",
|
|
name="my_tool",
|
|
arguments="{}",
|
|
type="function_call",
|
|
status="completed",
|
|
)
|
|
item = ToolCallItem(
|
|
agent=agent,
|
|
raw_item=raw_item,
|
|
title="My Tool",
|
|
description="A helpful tool",
|
|
)
|
|
|
|
result = item.to_input_item()
|
|
result_dict = cast(dict[str, Any], result)
|
|
|
|
assert isinstance(result, dict)
|
|
assert result_dict["type"] == "function_call"
|
|
assert "title" not in result_dict
|
|
assert "description" not in result_dict
|
|
|
|
|
|
def test_tool_call_item_tool_name_from_function_call() -> None:
|
|
"""ToolCallItem.tool_name should return the name attribute from a typed raw item."""
|
|
agent = Agent(name="test")
|
|
raw = ResponseFunctionToolCall(
|
|
id="fc1",
|
|
call_id="call_1",
|
|
name="my_tool",
|
|
arguments="{}",
|
|
type="function_call",
|
|
)
|
|
item = ToolCallItem(agent=agent, raw_item=raw)
|
|
assert item.tool_name == "my_tool"
|
|
|
|
|
|
def test_tool_call_item_tool_name_from_dict() -> None:
|
|
"""ToolCallItem.tool_name should return the 'name' key from a dict raw item."""
|
|
agent = Agent(name="test")
|
|
raw: dict[str, Any] = {
|
|
"type": "function_call",
|
|
"name": "dict_tool",
|
|
"call_id": "call_1",
|
|
"arguments": "{}",
|
|
}
|
|
item = ToolCallItem(agent=agent, raw_item=raw)
|
|
assert item.tool_name == "dict_tool"
|
|
|
|
|
|
def test_tool_call_item_tool_name_returns_none_when_missing() -> None:
|
|
"""ToolCallItem.tool_name should be None when the raw item has no name attribute."""
|
|
agent = Agent(name="test")
|
|
raw = ResponseFileSearchToolCall(
|
|
id="fs1",
|
|
queries=["q"],
|
|
status="completed",
|
|
type="file_search_call",
|
|
)
|
|
item = ToolCallItem(agent=agent, raw_item=raw)
|
|
assert item.tool_name is None
|
|
|
|
|
|
def test_tool_call_item_call_id_from_function_call() -> None:
|
|
"""ToolCallItem.call_id should return the call_id attribute from a typed raw item."""
|
|
agent = Agent(name="test")
|
|
raw = ResponseFunctionToolCall(
|
|
id="fc1",
|
|
call_id="call_abc",
|
|
name="t",
|
|
arguments="{}",
|
|
type="function_call",
|
|
)
|
|
item = ToolCallItem(agent=agent, raw_item=raw)
|
|
assert item.call_id == "call_abc"
|
|
|
|
|
|
def test_tool_call_item_call_id_falls_back_to_id() -> None:
|
|
"""ToolCallItem.call_id should fall back to id when call_id is absent."""
|
|
agent = Agent(name="test")
|
|
raw = ResponseFileSearchToolCall(
|
|
id="fs_xyz",
|
|
queries=["q"],
|
|
status="completed",
|
|
type="file_search_call",
|
|
)
|
|
item = ToolCallItem(agent=agent, raw_item=raw)
|
|
assert item.call_id == "fs_xyz"
|
|
|
|
|
|
def test_tool_call_item_call_id_from_dict() -> None:
|
|
"""ToolCallItem.call_id should return the 'call_id' key from a dict raw item."""
|
|
agent = Agent(name="test")
|
|
raw: dict[str, Any] = {
|
|
"type": "function_call",
|
|
"name": "t",
|
|
"call_id": "call_dict_id",
|
|
"arguments": "{}",
|
|
}
|
|
item = ToolCallItem(agent=agent, raw_item=raw)
|
|
assert item.call_id == "call_dict_id"
|
|
|
|
|
|
def test_tool_call_output_item_call_id_from_function_call_output() -> None:
|
|
"""ToolCallOutputItem.call_id should return call_id from the FunctionCallOutput dict."""
|
|
agent = Agent(name="test")
|
|
raw = {
|
|
"type": "function_call_output",
|
|
"call_id": "call_out_1",
|
|
"output": "ok",
|
|
}
|
|
item = ToolCallOutputItem(agent=agent, raw_item=raw, output="ok")
|
|
assert item.call_id == "call_out_1"
|
|
|
|
|
|
def test_tool_call_output_item_call_id_returns_none_when_missing() -> None:
|
|
"""ToolCallOutputItem.call_id should be None when neither call_id nor id are present."""
|
|
agent = Agent(name="test")
|
|
raw = {
|
|
"type": "function_call_output",
|
|
"output": "ok",
|
|
}
|
|
item = ToolCallOutputItem(agent=agent, raw_item=raw, output="ok")
|
|
assert item.call_id is None
|