wehub-resource-sync
1029 lines
33 KiB
Python
1029 lines
33 KiB
Python
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
from openai.types.responses import ResponseFunctionToolCall
|
|
from openai.types.responses.response_output_item import McpCall, McpListTools, McpListToolsTool
|
|
|
|
from agents import Agent, HostedMCPTool
|
|
from agents.items import (
|
|
MCPListToolsItem,
|
|
ModelResponse,
|
|
RunItem,
|
|
ToolApprovalItem,
|
|
ToolCallItem,
|
|
ToolCallOutputItem,
|
|
TResponseInputItem,
|
|
)
|
|
from agents.lifecycle import RunHooks
|
|
from agents.models.fake_id import FAKE_RESPONSES_ID
|
|
from agents.result import RunResultStreaming
|
|
from agents.run_config import ModelInputData, RunConfig
|
|
from agents.run_context import RunContextWrapper
|
|
from agents.run_internal.agent_bindings import bind_public_agent
|
|
from agents.run_internal.agent_runner_helpers import get_unsent_tool_call_ids_for_interrupted_state
|
|
from agents.run_internal.oai_conversation import OpenAIServerConversationTracker
|
|
from agents.run_internal.run_loop import get_new_response, run_single_turn_streamed
|
|
from agents.run_internal.run_steps import NextStepInterruption
|
|
from agents.run_internal.tool_use_tracker import AgentToolUseTracker
|
|
from agents.stream_events import RunItemStreamEvent
|
|
from agents.usage import Usage
|
|
|
|
from .fake_model import FakeModel
|
|
from .test_responses import get_text_message
|
|
|
|
|
|
class DummyRunItem:
|
|
"""Minimal stand-in for RunItem with the attributes used by OpenAIServerConversationTracker."""
|
|
|
|
def __init__(self, raw_item: dict[str, Any], type: str = "message") -> None:
|
|
self.raw_item = raw_item
|
|
self.type = type
|
|
|
|
|
|
def _make_hosted_mcp_list_tools(server_label: str, tool_name: str) -> McpListTools:
|
|
return McpListTools(
|
|
id=f"list_{server_label}",
|
|
server_label=server_label,
|
|
tools=[
|
|
McpListToolsTool(
|
|
name=tool_name,
|
|
input_schema={},
|
|
description="Search the docs.",
|
|
annotations={"title": "Search Docs"},
|
|
)
|
|
],
|
|
type="mcp_list_tools",
|
|
)
|
|
|
|
|
|
def test_prepare_input_filters_items_seen_by_server_and_tool_calls() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv", previous_response_id=None)
|
|
|
|
original_input: list[TResponseInputItem] = [
|
|
cast(TResponseInputItem, {"id": "input-1", "type": "message"}),
|
|
cast(TResponseInputItem, {"id": "input-2", "type": "message"}),
|
|
]
|
|
new_raw_item = {"type": "message", "content": "hello"}
|
|
generated_items = [
|
|
DummyRunItem({"id": "server-echo", "type": "message"}),
|
|
DummyRunItem(new_raw_item),
|
|
DummyRunItem({"call_id": "call-1", "output": "done"}, type="function_call_output_item"),
|
|
]
|
|
model_response = object.__new__(ModelResponse)
|
|
model_response.output = [
|
|
cast(Any, {"call_id": "call-1", "output": "prior", "type": "function_call_output"})
|
|
]
|
|
model_response.usage = Usage()
|
|
model_response.response_id = "resp-1"
|
|
session_items: list[TResponseInputItem] = [
|
|
cast(TResponseInputItem, {"id": "session-1", "type": "message"})
|
|
]
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input=original_input,
|
|
generated_items=cast(list[Any], generated_items),
|
|
model_responses=[model_response],
|
|
session_items=session_items,
|
|
)
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=original_input,
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
|
|
assert prepared == [new_raw_item]
|
|
assert tracker.sent_initial_input is True
|
|
assert tracker.remaining_initial_input is None
|
|
|
|
|
|
def test_hydrate_from_state_preserves_unsent_outputs_from_interrupted_turn() -> None:
|
|
agent = Agent(name="test")
|
|
cleanup1_call = ResponseFunctionToolCall(
|
|
id="fc_001",
|
|
type="function_call",
|
|
call_id="call_CLEANUP1",
|
|
name="run_cleanup",
|
|
arguments='{"target": "temp_files"}',
|
|
status="completed",
|
|
)
|
|
diagnostic_call = ResponseFunctionToolCall(
|
|
id="fc_002",
|
|
type="function_call",
|
|
call_id="call_DIAG",
|
|
name="run_diagnostic",
|
|
arguments='{"check_name": "thermal"}',
|
|
status="completed",
|
|
)
|
|
cleanup2_call = ResponseFunctionToolCall(
|
|
id="fc_003",
|
|
type="function_call",
|
|
call_id="call_CLEANUP2",
|
|
name="run_cleanup",
|
|
arguments='{"target": "winsxs_cache"}',
|
|
status="completed",
|
|
)
|
|
model_response = ModelResponse(
|
|
output=[cleanup1_call, diagnostic_call, cleanup2_call],
|
|
usage=Usage(),
|
|
response_id="resp_002",
|
|
)
|
|
diagnostic_output = ToolCallOutputItem(
|
|
agent=agent,
|
|
raw_item={
|
|
"type": "function_call_output",
|
|
"call_id": "call_DIAG",
|
|
"output": "Diagnostic completed.",
|
|
},
|
|
output="Diagnostic completed.",
|
|
)
|
|
generated_items: list[RunItem] = [
|
|
ToolCallItem(agent=agent, raw_item=cleanup1_call),
|
|
ToolCallItem(agent=agent, raw_item=diagnostic_call),
|
|
ToolCallItem(agent=agent, raw_item=cleanup2_call),
|
|
diagnostic_output,
|
|
ToolApprovalItem(agent=agent, raw_item=cleanup1_call, tool_name="run_cleanup"),
|
|
ToolApprovalItem(agent=agent, raw_item=cleanup2_call, tool_name="run_cleanup"),
|
|
]
|
|
interrupted_state = SimpleNamespace(
|
|
_current_step=NextStepInterruption(interruptions=[]),
|
|
_last_processed_response=SimpleNamespace(
|
|
handoffs=[],
|
|
functions=[
|
|
SimpleNamespace(tool_call=cleanup1_call),
|
|
SimpleNamespace(tool_call=diagnostic_call),
|
|
SimpleNamespace(tool_call=cleanup2_call),
|
|
],
|
|
computer_actions=[],
|
|
custom_tool_calls=[],
|
|
local_shell_calls=[],
|
|
shell_calls=[],
|
|
apply_patch_calls=[],
|
|
),
|
|
)
|
|
|
|
tracker = OpenAIServerConversationTracker(previous_response_id="resp_002")
|
|
tracker.hydrate_from_state(
|
|
original_input="Run cleanup, diagnostics, and cleanup.",
|
|
generated_items=generated_items,
|
|
model_responses=[model_response],
|
|
unsent_tool_call_ids=get_unsent_tool_call_ids_for_interrupted_state(
|
|
cast(Any, interrupted_state)
|
|
),
|
|
)
|
|
|
|
assert "call_DIAG" not in tracker.server_tool_call_ids
|
|
|
|
prepared = tracker.prepare_input(
|
|
"Run cleanup, diagnostics, and cleanup.",
|
|
[
|
|
ToolCallItem(agent=agent, raw_item=cleanup1_call),
|
|
ToolCallItem(agent=agent, raw_item=diagnostic_call),
|
|
ToolCallItem(agent=agent, raw_item=cleanup2_call),
|
|
diagnostic_output,
|
|
ToolCallOutputItem(
|
|
agent=agent,
|
|
raw_item={
|
|
"type": "function_call_output",
|
|
"call_id": "call_CLEANUP1",
|
|
"output": "Tool call not approved.",
|
|
},
|
|
output="Tool call not approved.",
|
|
),
|
|
ToolCallOutputItem(
|
|
agent=agent,
|
|
raw_item={
|
|
"type": "function_call_output",
|
|
"call_id": "call_CLEANUP2",
|
|
"output": "Tool call not approved.",
|
|
},
|
|
output="Tool call not approved.",
|
|
),
|
|
],
|
|
)
|
|
|
|
assert [
|
|
item.get("call_id")
|
|
for item in prepared
|
|
if isinstance(item, dict) and item.get("type") == "function_call_output"
|
|
] == ["call_DIAG", "call_CLEANUP1", "call_CLEANUP2"]
|
|
|
|
|
|
def test_hydrate_from_state_does_not_track_string_initial_input_by_object_identity() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv-init-string", previous_response_id=None
|
|
)
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input="hello",
|
|
generated_items=[],
|
|
model_responses=[],
|
|
)
|
|
|
|
assert tracker.sent_items == []
|
|
assert tracker.sent_initial_input is True
|
|
assert tracker.remaining_initial_input is None
|
|
assert len(tracker.sent_item_fingerprints) == 1
|
|
|
|
|
|
def test_hydrate_from_state_does_not_track_list_initial_input_by_object_identity() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv-init-list", previous_response_id=None
|
|
)
|
|
original_input = [cast(TResponseInputItem, {"role": "user", "content": "hello"})]
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input=original_input,
|
|
generated_items=[],
|
|
model_responses=[],
|
|
)
|
|
|
|
assert tracker.sent_items == []
|
|
assert tracker.sent_initial_input is True
|
|
assert tracker.remaining_initial_input is None
|
|
assert len(tracker.sent_item_fingerprints) == 1
|
|
|
|
|
|
def test_mark_input_as_sent_and_rewind_input_respects_remaining_initial_input() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv2", previous_response_id=None)
|
|
pending_1: TResponseInputItem = cast(TResponseInputItem, {"id": "p-1", "type": "message"})
|
|
pending_2: TResponseInputItem = cast(TResponseInputItem, {"id": "p-2", "type": "message"})
|
|
tracker.remaining_initial_input = [pending_1, pending_2]
|
|
|
|
tracker.mark_input_as_sent(
|
|
[pending_1, cast(TResponseInputItem, {"id": "p-2", "type": "message"})]
|
|
)
|
|
assert tracker.remaining_initial_input is None
|
|
|
|
tracker.rewind_input([pending_1])
|
|
assert tracker.remaining_initial_input == [pending_1]
|
|
|
|
|
|
def test_mark_input_as_sent_uses_raw_generated_source_for_rebuilt_filtered_item() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv2b", previous_response_id=None)
|
|
raw_generated_item = {
|
|
"type": "function_call_output",
|
|
"call_id": "call-2b",
|
|
"output": "done",
|
|
}
|
|
generated_items = [
|
|
DummyRunItem(raw_generated_item, type="function_call_output_item"),
|
|
]
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
rebuilt_filtered_item = cast(TResponseInputItem, dict(cast(dict[str, Any], prepared[0])))
|
|
|
|
tracker.mark_input_as_sent([rebuilt_filtered_item])
|
|
|
|
assert any(item is raw_generated_item for item in tracker.sent_items)
|
|
assert all(item is not rebuilt_filtered_item for item in tracker.sent_items)
|
|
|
|
prepared_again = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
assert prepared_again == []
|
|
|
|
|
|
def test_hydrate_from_state_skips_restored_tool_search_items_by_object_identity() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv2c", previous_response_id=None)
|
|
tool_search_call = {
|
|
"type": "tool_search_call",
|
|
"queries": [{"search_term": "account balance"}],
|
|
}
|
|
tool_search_result = {
|
|
"type": "tool_search_output",
|
|
"results": [{"text": "Balance lookup docs"}],
|
|
}
|
|
hydrated_items = [
|
|
DummyRunItem(tool_search_call, type="tool_search_call_item"),
|
|
DummyRunItem(tool_search_result, type="tool_search_output_item"),
|
|
]
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], hydrated_items),
|
|
model_responses=[],
|
|
)
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], hydrated_items),
|
|
)
|
|
|
|
assert prepared == []
|
|
|
|
|
|
def test_hydrate_from_state_skips_restored_tool_search_items_by_fingerprint() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv2d", previous_response_id=None)
|
|
tool_search_call = {
|
|
"type": "tool_search_call",
|
|
"queries": [{"search_term": "account balance"}],
|
|
}
|
|
tool_search_result = {
|
|
"type": "tool_search_output",
|
|
"results": [{"text": "Balance lookup docs"}],
|
|
}
|
|
hydrated_items = [
|
|
DummyRunItem(tool_search_call, type="tool_search_call_item"),
|
|
DummyRunItem(tool_search_result, type="tool_search_output_item"),
|
|
]
|
|
rebuilt_items = [
|
|
DummyRunItem(dict(tool_search_call), type="tool_search_call_item"),
|
|
DummyRunItem(dict(tool_search_result), type="tool_search_output_item"),
|
|
]
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], hydrated_items),
|
|
model_responses=[],
|
|
)
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], rebuilt_items),
|
|
)
|
|
|
|
assert prepared == []
|
|
|
|
|
|
def test_hydrate_from_state_skips_restored_tool_search_items_when_created_by_is_stripped() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv2d-created-by", previous_response_id=None
|
|
)
|
|
session_items = [
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_call",
|
|
"call_id": "tool_search_call_1",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"created_by": "server",
|
|
},
|
|
),
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_output",
|
|
"call_id": "tool_search_call_1",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
"created_by": "server",
|
|
},
|
|
),
|
|
]
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input=[],
|
|
generated_items=[],
|
|
model_responses=[],
|
|
session_items=session_items,
|
|
)
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(
|
|
list[RunItem],
|
|
[
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_call",
|
|
"call_id": "tool_search_call_1",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
type="tool_search_call_item",
|
|
),
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_output",
|
|
"call_id": "tool_search_call_1",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
},
|
|
type="tool_search_output_item",
|
|
),
|
|
],
|
|
),
|
|
)
|
|
|
|
assert prepared == []
|
|
|
|
|
|
def test_hydrate_from_state_skips_restored_tool_search_items_when_only_ids_differ() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv2d-ids-only", previous_response_id=None
|
|
)
|
|
session_items = [
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_call",
|
|
"id": "tool_search_call_saved",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
),
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_output",
|
|
"id": "tool_search_output_saved",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
},
|
|
),
|
|
]
|
|
|
|
tracker.hydrate_from_state(
|
|
original_input=[],
|
|
generated_items=[],
|
|
model_responses=[],
|
|
session_items=session_items,
|
|
)
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(
|
|
list[RunItem],
|
|
[
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_call",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
type="tool_search_call_item",
|
|
),
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_output",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
},
|
|
type="tool_search_output_item",
|
|
),
|
|
],
|
|
),
|
|
)
|
|
|
|
assert prepared == []
|
|
|
|
|
|
def test_prepare_input_keeps_repeated_tool_search_items_with_new_ids() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv2d-repeated-search", previous_response_id=None
|
|
)
|
|
|
|
prior_response = object.__new__(ModelResponse)
|
|
prior_response.output = [
|
|
cast(
|
|
Any,
|
|
{
|
|
"type": "tool_search_call",
|
|
"id": "tool_search_call_saved",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"created_by": "server",
|
|
},
|
|
),
|
|
cast(
|
|
Any,
|
|
{
|
|
"type": "tool_search_output",
|
|
"id": "tool_search_output_saved",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
"created_by": "server",
|
|
},
|
|
),
|
|
]
|
|
prior_response.usage = Usage()
|
|
prior_response.response_id = "resp-tool-search-repeat-1"
|
|
|
|
tracker.track_server_items(prior_response)
|
|
|
|
repeated_items = [
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_call",
|
|
"id": "tool_search_call_repeat",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
type="tool_search_call_item",
|
|
),
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_output",
|
|
"id": "tool_search_output_repeat",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
},
|
|
type="tool_search_output_item",
|
|
),
|
|
]
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], repeated_items),
|
|
)
|
|
|
|
assert prepared == [
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_call",
|
|
"id": "tool_search_call_repeat",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
),
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_output",
|
|
"id": "tool_search_output_repeat",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
},
|
|
),
|
|
]
|
|
|
|
|
|
def test_track_server_items_skips_live_tool_search_items_on_next_prepare() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv2e", previous_response_id=None)
|
|
tool_search_call = cast(
|
|
Any,
|
|
{
|
|
"type": "tool_search_call",
|
|
"call_id": "tool_search_call_live",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"created_by": "server",
|
|
},
|
|
)
|
|
tool_search_result = cast(
|
|
Any,
|
|
{
|
|
"type": "tool_search_output",
|
|
"call_id": "tool_search_call_live",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
"created_by": "server",
|
|
},
|
|
)
|
|
model_response = object.__new__(ModelResponse)
|
|
model_response.output = [tool_search_call, tool_search_result]
|
|
model_response.usage = Usage()
|
|
model_response.response_id = "resp-tool-search"
|
|
|
|
tracker.track_server_items(model_response)
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(
|
|
list[RunItem],
|
|
[
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_call",
|
|
"call_id": "tool_search_call_live",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
type="tool_search_call_item",
|
|
),
|
|
DummyRunItem(
|
|
{
|
|
"type": "tool_search_output",
|
|
"call_id": "tool_search_call_live",
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"tools": [],
|
|
},
|
|
type="tool_search_output_item",
|
|
),
|
|
],
|
|
),
|
|
)
|
|
|
|
assert prepared == []
|
|
|
|
|
|
def test_track_server_items_filters_pending_tool_search_by_sanitized_fingerprint() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv2e-pending", previous_response_id=None
|
|
)
|
|
tracker.remaining_initial_input = [
|
|
cast(
|
|
TResponseInputItem,
|
|
{
|
|
"type": "tool_search_call",
|
|
"call_id": "tool_search_pending",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
},
|
|
),
|
|
cast(TResponseInputItem, {"id": "keep-me", "type": "message"}),
|
|
]
|
|
|
|
model_response = object.__new__(ModelResponse)
|
|
model_response.output = [
|
|
cast(
|
|
Any,
|
|
{
|
|
"type": "tool_search_call",
|
|
"call_id": "tool_search_pending",
|
|
"arguments": {"query": "account balance"},
|
|
"execution": "server",
|
|
"status": "completed",
|
|
"created_by": "server",
|
|
},
|
|
)
|
|
]
|
|
model_response.usage = Usage()
|
|
model_response.response_id = "resp-tool-search-pending"
|
|
|
|
tracker.track_server_items(model_response)
|
|
|
|
assert tracker.remaining_initial_input == [
|
|
cast(TResponseInputItem, {"id": "keep-me", "type": "message"})
|
|
]
|
|
|
|
|
|
def test_track_server_items_filters_remaining_initial_input_by_fingerprint() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv3", previous_response_id=None)
|
|
pending_kept: TResponseInputItem = cast(
|
|
TResponseInputItem, {"id": "keep-me", "type": "message"}
|
|
)
|
|
pending_filtered: TResponseInputItem = cast(
|
|
TResponseInputItem,
|
|
{"type": "function_call_output", "call_id": "call-2", "output": "x"},
|
|
)
|
|
tracker.remaining_initial_input = [pending_kept, pending_filtered]
|
|
|
|
model_response = object.__new__(ModelResponse)
|
|
model_response.output = [
|
|
cast(Any, {"type": "function_call_output", "call_id": "call-2", "output": "x"})
|
|
]
|
|
model_response.usage = Usage()
|
|
model_response.response_id = "resp-2"
|
|
|
|
tracker.track_server_items(model_response)
|
|
|
|
assert tracker.remaining_initial_input == [pending_kept]
|
|
|
|
|
|
def test_prepare_input_does_not_skip_fake_response_ids() -> None:
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv5", previous_response_id=None)
|
|
|
|
model_response = object.__new__(ModelResponse)
|
|
model_response.output = [cast(Any, {"id": FAKE_RESPONSES_ID, "type": "message"})]
|
|
model_response.usage = Usage()
|
|
model_response.response_id = "resp-3"
|
|
|
|
tracker.track_server_items(model_response)
|
|
|
|
raw_item = {"id": FAKE_RESPONSES_ID, "type": "message", "content": "hello"}
|
|
generated_items = [DummyRunItem(raw_item)]
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
|
|
assert prepared == [raw_item]
|
|
|
|
|
|
def test_prepare_input_applies_reasoning_item_id_policy_for_generated_items() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv7",
|
|
previous_response_id=None,
|
|
reasoning_item_id_policy="omit",
|
|
)
|
|
generated_items = [
|
|
DummyRunItem(
|
|
{
|
|
"type": "reasoning",
|
|
"id": "rs_turn_input",
|
|
"content": [{"type": "input_text", "text": "reasoning trace"}],
|
|
},
|
|
type="reasoning_item",
|
|
)
|
|
]
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
|
|
assert prepared == [
|
|
cast(
|
|
TResponseInputItem,
|
|
{"type": "reasoning", "content": [{"type": "input_text", "text": "reasoning trace"}]},
|
|
)
|
|
]
|
|
|
|
|
|
def test_prepare_input_does_not_resend_reasoning_item_after_marking_omitted_id_as_sent() -> None:
|
|
tracker = OpenAIServerConversationTracker(
|
|
conversation_id="conv8",
|
|
previous_response_id=None,
|
|
reasoning_item_id_policy="omit",
|
|
)
|
|
generated_items = [
|
|
DummyRunItem(
|
|
{
|
|
"type": "reasoning",
|
|
"id": "rs_turn_input",
|
|
"content": [{"type": "input_text", "text": "reasoning trace"}],
|
|
},
|
|
type="reasoning_item",
|
|
)
|
|
]
|
|
|
|
first_prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
assert first_prepared == [
|
|
cast(
|
|
TResponseInputItem,
|
|
{"type": "reasoning", "content": [{"type": "input_text", "text": "reasoning trace"}]},
|
|
)
|
|
]
|
|
|
|
tracker.mark_input_as_sent(first_prepared)
|
|
|
|
second_prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
assert second_prepared == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_new_response_marks_filtered_input_as_sent() -> None:
|
|
model = FakeModel()
|
|
model.set_next_output([get_text_message("ok")])
|
|
agent = Agent(name="test", model=model)
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv4", previous_response_id=None)
|
|
context_wrapper: RunContextWrapper[dict[str, Any]] = RunContextWrapper(context={})
|
|
tool_use_tracker = AgentToolUseTracker()
|
|
|
|
item_1: TResponseInputItem = cast(TResponseInputItem, {"role": "user", "content": "first"})
|
|
item_2: TResponseInputItem = cast(TResponseInputItem, {"role": "user", "content": "second"})
|
|
|
|
def _filter_input(payload: Any) -> ModelInputData:
|
|
return ModelInputData(
|
|
input=[payload.model_data.input[0]],
|
|
instructions=payload.model_data.instructions,
|
|
)
|
|
|
|
run_config = RunConfig(call_model_input_filter=_filter_input)
|
|
|
|
await get_new_response(
|
|
bind_public_agent(agent),
|
|
None,
|
|
[item_1, item_2],
|
|
None,
|
|
[],
|
|
[],
|
|
RunHooks(),
|
|
context_wrapper,
|
|
run_config,
|
|
tool_use_tracker,
|
|
tracker,
|
|
None,
|
|
)
|
|
|
|
assert model.last_turn_args["input"] == [item_1]
|
|
assert any(item is item_1 for item in tracker.sent_items)
|
|
assert all(item is not item_2 for item in tracker.sent_items)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_single_turn_streamed_marks_filtered_input_as_sent() -> None:
|
|
model = FakeModel()
|
|
model.set_next_output([get_text_message("ok")])
|
|
agent = Agent(name="test", model=model)
|
|
tracker = OpenAIServerConversationTracker(conversation_id="conv6", previous_response_id=None)
|
|
context_wrapper: RunContextWrapper[dict[str, Any]] = RunContextWrapper(context={})
|
|
tool_use_tracker = AgentToolUseTracker()
|
|
|
|
item_1: TResponseInputItem = cast(TResponseInputItem, {"role": "user", "content": "first"})
|
|
item_2: TResponseInputItem = cast(TResponseInputItem, {"role": "user", "content": "second"})
|
|
|
|
def _filter_input(payload: Any) -> ModelInputData:
|
|
return ModelInputData(
|
|
input=[payload.model_data.input[0]],
|
|
instructions=payload.model_data.instructions,
|
|
)
|
|
|
|
run_config = RunConfig(call_model_input_filter=_filter_input)
|
|
|
|
streamed_result = RunResultStreaming(
|
|
input=[item_1, item_2],
|
|
new_items=[],
|
|
raw_responses=[],
|
|
final_output=None,
|
|
input_guardrail_results=[],
|
|
output_guardrail_results=[],
|
|
tool_input_guardrail_results=[],
|
|
tool_output_guardrail_results=[],
|
|
context_wrapper=context_wrapper,
|
|
current_agent=agent,
|
|
current_turn=0,
|
|
max_turns=1,
|
|
_current_agent_output_schema=None,
|
|
trace=None,
|
|
interruptions=[],
|
|
)
|
|
|
|
await run_single_turn_streamed(
|
|
streamed_result,
|
|
bind_public_agent(agent),
|
|
RunHooks(),
|
|
context_wrapper,
|
|
run_config,
|
|
should_run_agent_start_hooks=False,
|
|
tool_use_tracker=tool_use_tracker,
|
|
all_tools=[],
|
|
server_conversation_tracker=tracker,
|
|
)
|
|
|
|
assert model.last_turn_args["input"] == [item_1]
|
|
assert tracker.remaining_initial_input == [item_2]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_single_turn_streamed_seeds_hosted_mcp_metadata_from_pre_step_items() -> None:
|
|
model = FakeModel()
|
|
mcp_call = McpCall(
|
|
id="mcp_call_1",
|
|
arguments="{}",
|
|
name="search_docs",
|
|
server_label="docs_server",
|
|
type="mcp_call",
|
|
status="completed",
|
|
)
|
|
model.set_next_output([mcp_call])
|
|
agent = Agent(name="test", model=model)
|
|
hosted_tool = HostedMCPTool(
|
|
tool_config=cast(
|
|
Any,
|
|
{
|
|
"type": "mcp",
|
|
"server_label": "docs_server",
|
|
"server_url": "https://example.com/mcp",
|
|
},
|
|
)
|
|
)
|
|
context_wrapper: RunContextWrapper[dict[str, Any]] = RunContextWrapper(context={})
|
|
tool_use_tracker = AgentToolUseTracker()
|
|
|
|
item_1: TResponseInputItem = cast(TResponseInputItem, {"role": "user", "content": "first"})
|
|
pre_step_item = MCPListToolsItem(
|
|
agent=agent,
|
|
raw_item=_make_hosted_mcp_list_tools("docs_server", "search_docs"),
|
|
)
|
|
|
|
def _filter_input(payload: Any) -> ModelInputData:
|
|
return ModelInputData(
|
|
input=[payload.model_data.input[0]],
|
|
instructions=payload.model_data.instructions,
|
|
)
|
|
|
|
run_config = RunConfig(call_model_input_filter=_filter_input)
|
|
|
|
streamed_result = RunResultStreaming(
|
|
input=[item_1],
|
|
new_items=[],
|
|
raw_responses=[],
|
|
final_output=None,
|
|
input_guardrail_results=[],
|
|
output_guardrail_results=[],
|
|
tool_input_guardrail_results=[],
|
|
tool_output_guardrail_results=[],
|
|
context_wrapper=context_wrapper,
|
|
current_agent=agent,
|
|
current_turn=1,
|
|
max_turns=2,
|
|
_current_agent_output_schema=None,
|
|
trace=None,
|
|
interruptions=[],
|
|
)
|
|
streamed_result._model_input_items = [pre_step_item]
|
|
|
|
await run_single_turn_streamed(
|
|
streamed_result,
|
|
bind_public_agent(agent),
|
|
RunHooks(),
|
|
context_wrapper,
|
|
run_config,
|
|
should_run_agent_start_hooks=False,
|
|
tool_use_tracker=tool_use_tracker,
|
|
all_tools=[hosted_tool],
|
|
)
|
|
|
|
assert model.last_turn_args["input"] == [item_1]
|
|
|
|
tool_call_events: list[ToolCallItem] = []
|
|
while not streamed_result._event_queue.empty():
|
|
queued_event = streamed_result._event_queue.get_nowait()
|
|
streamed_result._event_queue.task_done()
|
|
if (
|
|
isinstance(queued_event, RunItemStreamEvent)
|
|
and queued_event.name == "tool_called"
|
|
and isinstance(queued_event.item, ToolCallItem)
|
|
):
|
|
tool_call_events.append(queued_event.item)
|
|
|
|
assert len(tool_call_events) == 1
|
|
assert tool_call_events[0].description == "Search the docs."
|
|
assert tool_call_events[0].title == "Search Docs"
|
|
|
|
|
|
@pytest.mark.parametrize("stale_collection_name", ["sent_items", "server_items"])
|
|
def test_prepare_input_keeps_fresh_tool_output_when_stale_identity_matches(
|
|
stale_collection_name: str,
|
|
) -> None:
|
|
"""Tracked object identity must not become a stale address-based dedupe key."""
|
|
tracker = OpenAIServerConversationTracker(previous_response_id="resp-1")
|
|
|
|
output_raw_item: dict[str, Any] = {
|
|
"type": "function_call_output",
|
|
"call_id": "call_FRESH",
|
|
"output": "42",
|
|
}
|
|
tracked_items = getattr(tracker, stale_collection_name)
|
|
if isinstance(tracked_items, set):
|
|
tracked_items.add(id(output_raw_item))
|
|
else:
|
|
old_item = {"type": "message", "content": "already tracked"}
|
|
tracked_items.append(old_item)
|
|
|
|
generated_items = [DummyRunItem(output_raw_item, type="function_call_output_item")]
|
|
|
|
prepared = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
|
|
prepared_output_call_ids = [
|
|
item.get("call_id")
|
|
for item in prepared
|
|
if isinstance(item, dict) and item.get("type") == "function_call_output"
|
|
]
|
|
assert "call_FRESH" in prepared_output_call_ids
|
|
|
|
|
|
def test_prepare_input_dedupes_same_delivered_tool_output_object() -> None:
|
|
"""Identity dedupe still skips the exact source object after it is delivered."""
|
|
tracker = OpenAIServerConversationTracker(previous_response_id="resp-1")
|
|
|
|
output_raw_item: dict[str, Any] = {
|
|
"type": "function_call_output",
|
|
"call_id": "call_X",
|
|
"output": "42",
|
|
}
|
|
generated_items = [DummyRunItem(output_raw_item, type="function_call_output_item")]
|
|
|
|
first = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
assert any(isinstance(item, dict) and item.get("call_id") == "call_X" for item in first)
|
|
|
|
tracker.mark_input_as_sent(first)
|
|
assert any(item is output_raw_item for item in tracker.sent_items)
|
|
|
|
second = tracker.prepare_input(
|
|
original_input=[],
|
|
generated_items=cast(list[Any], generated_items),
|
|
)
|
|
assert all(not (isinstance(item, dict) and item.get("call_id") == "call_X") for item in second)
|