wehub-resource-sync
787 lines
28 KiB
Python
787 lines
28 KiB
Python
"""Unit tests for the ComputerAction methods in `agents.run_internal.run_loop`.
|
|
|
|
These confirm that the correct computer action method is invoked for each action type and
|
|
that screenshots are taken and wrapped appropriately, and that the execute function invokes
|
|
hooks and returns the expected ToolCallOutputItem."""
|
|
|
|
import json
|
|
import logging
|
|
from collections.abc import Callable
|
|
from typing import Any, TypeVar, cast
|
|
|
|
import pytest
|
|
from openai.types.responses.computer_action import (
|
|
Click as BatchedClick,
|
|
Screenshot as BatchedScreenshot,
|
|
Type as BatchedType,
|
|
)
|
|
from openai.types.responses.response_computer_tool_call import (
|
|
ActionClick,
|
|
ActionDoubleClick,
|
|
ActionDrag,
|
|
ActionDragPath,
|
|
ActionKeypress,
|
|
ActionMove,
|
|
ActionScreenshot,
|
|
ActionScroll,
|
|
ActionType,
|
|
ActionWait,
|
|
PendingSafetyCheck,
|
|
ResponseComputerToolCall,
|
|
)
|
|
|
|
from agents import (
|
|
Agent,
|
|
AgentHooks,
|
|
AsyncComputer,
|
|
Computer,
|
|
ComputerTool,
|
|
RunConfig,
|
|
RunContextWrapper,
|
|
RunHooks,
|
|
Runner,
|
|
set_tracing_disabled,
|
|
trace,
|
|
)
|
|
from agents.items import ToolCallOutputItem
|
|
from agents.run_internal import run_loop
|
|
from agents.run_internal.run_loop import ComputerAction, ToolRunComputerAction
|
|
from agents.tool import ComputerToolSafetyCheckData
|
|
|
|
from .fake_model import FakeModel
|
|
from .test_responses import get_text_message
|
|
from .testing_processor import SPAN_PROCESSOR_TESTING
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
def _get_function_span(tool_name: str) -> dict[str, Any]:
|
|
for span in SPAN_PROCESSOR_TESTING.get_ordered_spans(including_empty=True):
|
|
exported = span.export()
|
|
if not exported:
|
|
continue
|
|
span_data = exported.get("span_data")
|
|
if not isinstance(span_data, dict):
|
|
continue
|
|
if span_data.get("type") == "function" and span_data.get("name") == tool_name:
|
|
return exported
|
|
raise AssertionError(f"Function span for tool '{tool_name}' not found")
|
|
|
|
|
|
def _get_agent_span(agent_name: str) -> dict[str, Any]:
|
|
for span in SPAN_PROCESSOR_TESTING.get_ordered_spans(including_empty=True):
|
|
exported = span.export()
|
|
if not exported:
|
|
continue
|
|
span_data = exported.get("span_data")
|
|
if not isinstance(span_data, dict):
|
|
continue
|
|
if span_data.get("type") == "agent" and span_data.get("name") == agent_name:
|
|
return exported
|
|
raise AssertionError(f"Agent span for '{agent_name}' not found")
|
|
|
|
|
|
def _action_with_keys(factory: Callable[..., T], **kwargs: Any) -> T:
|
|
return cast(T, cast(Any, factory)(**kwargs))
|
|
|
|
|
|
class LoggingComputer(Computer):
|
|
"""A `Computer` implementation that logs calls to its methods for verification in tests."""
|
|
|
|
def __init__(self, screenshot_return: str = "screenshot"):
|
|
self.calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
self._screenshot_return = screenshot_return
|
|
|
|
@property
|
|
def environment(self):
|
|
return "mac"
|
|
|
|
@property
|
|
def dimensions(self) -> tuple[int, int]:
|
|
return (800, 600)
|
|
|
|
def screenshot(self) -> str:
|
|
self.calls.append(("screenshot", ()))
|
|
return self._screenshot_return
|
|
|
|
def _log_mouse_action(self, name: str, *args: Any, keys: list[str] | None = None) -> None:
|
|
payload = args if keys is None else (*args, keys)
|
|
self.calls.append((name, payload))
|
|
|
|
def click(self, x: int, y: int, button: str, *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("click", x, y, button, keys=keys)
|
|
|
|
def double_click(self, x: int, y: int, *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("double_click", x, y, keys=keys)
|
|
|
|
def scroll(
|
|
self, x: int, y: int, scroll_x: int, scroll_y: int, *, keys: list[str] | None = None
|
|
) -> None:
|
|
self._log_mouse_action("scroll", x, y, scroll_x, scroll_y, keys=keys)
|
|
|
|
def type(self, text: str) -> None:
|
|
self.calls.append(("type", (text,)))
|
|
|
|
def wait(self) -> None:
|
|
self.calls.append(("wait", ()))
|
|
|
|
def move(self, x: int, y: int, *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("move", x, y, keys=keys)
|
|
|
|
def keypress(self, keys: list[str]) -> None:
|
|
self.calls.append(("keypress", (keys,)))
|
|
|
|
def drag(self, path: list[tuple[int, int]], *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("drag", tuple(path), keys=keys)
|
|
|
|
|
|
class LoggingAsyncComputer(AsyncComputer):
|
|
"""An `AsyncComputer` implementation that logs calls to its methods for verification."""
|
|
|
|
def __init__(self, screenshot_return: str = "async_screenshot"):
|
|
self.calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
self._screenshot_return = screenshot_return
|
|
|
|
@property
|
|
def environment(self):
|
|
return "mac"
|
|
|
|
@property
|
|
def dimensions(self) -> tuple[int, int]:
|
|
return (800, 600)
|
|
|
|
async def screenshot(self) -> str:
|
|
self.calls.append(("screenshot", ()))
|
|
return self._screenshot_return
|
|
|
|
def _log_mouse_action(self, name: str, *args: Any, keys: list[str] | None = None) -> None:
|
|
payload = args if keys is None else (*args, keys)
|
|
self.calls.append((name, payload))
|
|
|
|
async def click(self, x: int, y: int, button: str, *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("click", x, y, button, keys=keys)
|
|
|
|
async def double_click(self, x: int, y: int, *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("double_click", x, y, keys=keys)
|
|
|
|
async def scroll(
|
|
self, x: int, y: int, scroll_x: int, scroll_y: int, *, keys: list[str] | None = None
|
|
) -> None:
|
|
self._log_mouse_action("scroll", x, y, scroll_x, scroll_y, keys=keys)
|
|
|
|
async def type(self, text: str) -> None:
|
|
self.calls.append(("type", (text,)))
|
|
|
|
async def wait(self) -> None:
|
|
self.calls.append(("wait", ()))
|
|
|
|
async def move(self, x: int, y: int, *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("move", x, y, keys=keys)
|
|
|
|
async def keypress(self, keys: list[str]) -> None:
|
|
self.calls.append(("keypress", (keys,)))
|
|
|
|
async def drag(self, path: list[tuple[int, int]], *, keys: list[str] | None = None) -> None:
|
|
self._log_mouse_action("drag", tuple(path), keys=keys)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"action,expected_call",
|
|
[
|
|
(ActionClick(type="click", x=10, y=21, button="left"), ("click", (10, 21, "left"))),
|
|
(ActionDoubleClick(type="double_click", x=42, y=47), ("double_click", (42, 47))),
|
|
(
|
|
ActionDrag(type="drag", path=[ActionDragPath(x=1, y=2), ActionDragPath(x=3, y=4)]),
|
|
("drag", (((1, 2), (3, 4)),)),
|
|
),
|
|
(ActionKeypress(type="keypress", keys=["a", "b"]), ("keypress", (["a", "b"],))),
|
|
(ActionMove(type="move", x=100, y=200), ("move", (100, 200))),
|
|
(ActionScreenshot(type="screenshot"), ("screenshot", ())),
|
|
(
|
|
ActionScroll(type="scroll", x=1, y=2, scroll_x=3, scroll_y=4),
|
|
("scroll", (1, 2, 3, 4)),
|
|
),
|
|
(ActionType(type="type", text="hello"), ("type", ("hello",))),
|
|
(ActionWait(type="wait"), ("wait", ())),
|
|
],
|
|
)
|
|
async def test_get_screenshot_sync_executes_action_and_takes_screenshot(
|
|
action: Any, expected_call: tuple[str, tuple[Any, ...]]
|
|
) -> None:
|
|
"""For each action type, assert that the corresponding computer method is invoked
|
|
and that a screenshot is taken and returned."""
|
|
computer = LoggingComputer(screenshot_return="synthetic")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c1",
|
|
type="computer_call",
|
|
action=action,
|
|
call_id="c1",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
if isinstance(action, ActionScreenshot):
|
|
assert computer.calls == [("screenshot", ())]
|
|
else:
|
|
assert computer.calls == [expected_call, ("screenshot", ())]
|
|
assert screenshot_output == "synthetic"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"action,expected_call",
|
|
[
|
|
(ActionClick(type="click", x=2, y=3, button="right"), ("click", (2, 3, "right"))),
|
|
(ActionDoubleClick(type="double_click", x=12, y=13), ("double_click", (12, 13))),
|
|
(
|
|
ActionDrag(type="drag", path=[ActionDragPath(x=5, y=6), ActionDragPath(x=6, y=7)]),
|
|
("drag", (((5, 6), (6, 7)),)),
|
|
),
|
|
(ActionKeypress(type="keypress", keys=["ctrl", "c"]), ("keypress", (["ctrl", "c"],))),
|
|
(ActionMove(type="move", x=8, y=9), ("move", (8, 9))),
|
|
(ActionScreenshot(type="screenshot"), ("screenshot", ())),
|
|
(
|
|
ActionScroll(type="scroll", x=9, y=8, scroll_x=7, scroll_y=6),
|
|
("scroll", (9, 8, 7, 6)),
|
|
),
|
|
(ActionType(type="type", text="world"), ("type", ("world",))),
|
|
(ActionWait(type="wait"), ("wait", ())),
|
|
],
|
|
)
|
|
async def test_get_screenshot_async_executes_action_and_takes_screenshot(
|
|
action: Any, expected_call: tuple[str, tuple[Any, ...]]
|
|
) -> None:
|
|
"""For each action type on an `AsyncComputer`, the corresponding coroutine should be awaited
|
|
and a screenshot taken."""
|
|
computer = LoggingAsyncComputer(screenshot_return="async_return")
|
|
assert computer.environment == "mac"
|
|
assert computer.dimensions == (800, 600)
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c2",
|
|
type="computer_call",
|
|
action=action,
|
|
call_id="c2",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
if isinstance(action, ActionScreenshot):
|
|
assert computer.calls == [("screenshot", ())]
|
|
else:
|
|
assert computer.calls == [expected_call, ("screenshot", ())]
|
|
assert screenshot_output == "async_return"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_executes_batched_actions_in_order() -> None:
|
|
computer = LoggingComputer(screenshot_return="batched")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c3",
|
|
type="computer_call",
|
|
actions=[
|
|
BatchedClick(type="click", x=11, y=12, button="left"),
|
|
BatchedType(type="type", text="hello"),
|
|
],
|
|
call_id="c3",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
|
|
assert computer.calls == [
|
|
("click", (11, 12, "left")),
|
|
("type", ("hello",)),
|
|
("screenshot", ()),
|
|
]
|
|
assert screenshot_output == "batched"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_reuses_terminal_batched_screenshot() -> None:
|
|
computer = LoggingComputer(screenshot_return="captured")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c4",
|
|
type="computer_call",
|
|
actions=[BatchedScreenshot(type="screenshot")],
|
|
call_id="c4",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
|
|
assert computer.calls == [("screenshot", ())]
|
|
assert screenshot_output == "captured"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_preserves_modifier_keys_for_sync_driver() -> None:
|
|
computer = LoggingComputer(screenshot_return="with_keys")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c5",
|
|
type="computer_call",
|
|
action=_action_with_keys(
|
|
ActionClick, type="click", x=4, y=8, button="left", keys=["shift", "ctrl"]
|
|
),
|
|
call_id="c5",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
|
|
assert computer.calls == [
|
|
("click", (4, 8, "left", ["shift", "ctrl"])),
|
|
("screenshot", ()),
|
|
]
|
|
assert screenshot_output == "with_keys"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_preserves_modifier_keys_for_async_driver() -> None:
|
|
computer = LoggingAsyncComputer(screenshot_return="async_keys")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c6",
|
|
type="computer_call",
|
|
action=_action_with_keys(
|
|
ActionScroll, type="scroll", x=7, y=9, scroll_x=3, scroll_y=-2, keys=["alt"]
|
|
),
|
|
call_id="c6",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
|
|
assert computer.calls == [
|
|
("scroll", (7, 9, 3, -2, ["alt"])),
|
|
("screenshot", ()),
|
|
]
|
|
assert screenshot_output == "async_keys"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_drops_modifier_keys_for_legacy_driver_with_warning(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
class LegacyDriver:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
|
|
def screenshot(self) -> str:
|
|
self.calls.append(("screenshot", ()))
|
|
return "legacy"
|
|
|
|
def click(self, x: int, y: int, button: str) -> None:
|
|
self.calls.append(("click", (x, y, button)))
|
|
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c7",
|
|
type="computer_call",
|
|
action=_action_with_keys(
|
|
ActionClick, type="click", x=1, y=1, button="left", keys=["shift"]
|
|
),
|
|
call_id="c7",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
driver = LegacyDriver()
|
|
with caplog.at_level(logging.WARNING, logger="openai.agents"):
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(driver, tool_call)
|
|
|
|
assert driver.calls == [("click", (1, 1, "left")), ("screenshot", ())]
|
|
assert screenshot_output == "legacy"
|
|
assert "does not accept keyword argument(s) keys" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_drops_modifier_keys_for_non_introspectable_driver_with_warning(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
class NonIntrospectableClick:
|
|
def __init__(self, calls: list[tuple[str, tuple[Any, ...]]]) -> None:
|
|
self._calls = calls
|
|
|
|
@property
|
|
def __signature__(self) -> Any:
|
|
raise ValueError("signature unavailable")
|
|
|
|
def __call__(self, x: int, y: int, button: str) -> None:
|
|
self._calls.append(("click", (x, y, button)))
|
|
|
|
class NonIntrospectableDriver:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, tuple[Any, ...]]] = []
|
|
self.click = NonIntrospectableClick(self.calls)
|
|
|
|
def screenshot(self) -> str:
|
|
self.calls.append(("screenshot", ()))
|
|
return "non_introspectable"
|
|
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c8",
|
|
type="computer_call",
|
|
action=_action_with_keys(
|
|
ActionClick, type="click", x=2, y=5, button="left", keys=["shift"]
|
|
),
|
|
call_id="c8",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
driver = NonIntrospectableDriver()
|
|
with caplog.at_level(logging.WARNING, logger="openai.agents"):
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(driver, tool_call)
|
|
|
|
assert driver.calls == [("click", (2, 5, "left")), ("screenshot", ())]
|
|
assert screenshot_output == "non_introspectable"
|
|
assert "does not accept keyword argument(s) keys" in caplog.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_preserves_modifier_keys_for_kwargs_driver() -> None:
|
|
class KwargsDriver:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = []
|
|
|
|
def screenshot(self) -> str:
|
|
self.calls.append(("screenshot", (), {}))
|
|
return "kwargs"
|
|
|
|
def move(self, x: int, y: int, **kwargs: Any) -> None:
|
|
self.calls.append(("move", (x, y), kwargs))
|
|
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c9",
|
|
type="computer_call",
|
|
action=_action_with_keys(ActionMove, type="move", x=10, y=12, keys=["meta"]),
|
|
call_id="c9",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
driver = KwargsDriver()
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(driver, tool_call)
|
|
|
|
assert driver.calls == [
|
|
("move", (10, 12), {"keys": ["meta"]}),
|
|
("screenshot", (), {}),
|
|
]
|
|
assert screenshot_output == "kwargs"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_screenshot_preserves_modifier_keys_for_batched_actions() -> None:
|
|
computer = LoggingComputer(screenshot_return="batched_keys")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="c10",
|
|
type="computer_call",
|
|
actions=[
|
|
_action_with_keys(BatchedClick, type="click", x=11, y=12, button="left", keys=["ctrl"])
|
|
],
|
|
call_id="c10",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
|
|
screenshot_output = await ComputerAction._execute_action_and_capture(computer, tool_call)
|
|
|
|
assert computer.calls == [
|
|
("click", (11, 12, "left", ["ctrl"])),
|
|
("screenshot", ()),
|
|
]
|
|
assert screenshot_output == "batched_keys"
|
|
|
|
|
|
class LoggingRunHooks(RunHooks[Any]):
|
|
"""Capture on_tool_start and on_tool_end invocations."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.started: list[tuple[Agent[Any], Any]] = []
|
|
self.ended: list[tuple[Agent[Any], Any, object]] = []
|
|
|
|
async def on_tool_start(
|
|
self, context: RunContextWrapper[Any], agent: Agent[Any], tool: Any
|
|
) -> None:
|
|
self.started.append((agent, tool))
|
|
|
|
async def on_tool_end(
|
|
self, context: RunContextWrapper[Any], agent: Agent[Any], tool: Any, result: object
|
|
) -> None:
|
|
self.ended.append((agent, tool, result))
|
|
|
|
|
|
class LoggingAgentHooks(AgentHooks[Any]):
|
|
"""Minimal override to capture agent's tool hook invocations."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.started: list[tuple[Agent[Any], Any]] = []
|
|
self.ended: list[tuple[Agent[Any], Any, object]] = []
|
|
|
|
async def on_tool_start(
|
|
self, context: RunContextWrapper[Any], agent: Agent[Any], tool: Any
|
|
) -> None:
|
|
self.started.append((agent, tool))
|
|
|
|
async def on_tool_end(
|
|
self, context: RunContextWrapper[Any], agent: Agent[Any], tool: Any, result: object
|
|
) -> None:
|
|
self.ended.append((agent, tool, result))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_invokes_hooks_and_returns_tool_call_output() -> None:
|
|
# ComputerAction.execute should invoke lifecycle hooks and return a proper ToolCallOutputItem.
|
|
computer = LoggingComputer(screenshot_return="xyz")
|
|
comptool = ComputerTool(computer=computer)
|
|
# Create a dummy click action to trigger a click and screenshot.
|
|
action = ActionClick(type="click", x=1, y=2, button="left")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="tool123",
|
|
type="computer_call",
|
|
action=action,
|
|
call_id="tool123",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
tool_call.call_id = "tool123"
|
|
|
|
# Wrap tool call in ToolRunComputerAction
|
|
tool_run = ToolRunComputerAction(tool_call=tool_call, computer_tool=comptool)
|
|
# Setup agent and hooks.
|
|
agent = Agent(name="test_agent", tools=[comptool])
|
|
# Attach per-agent hooks as well as global run hooks.
|
|
agent_hooks = LoggingAgentHooks()
|
|
agent.hooks = agent_hooks
|
|
run_hooks = LoggingRunHooks()
|
|
context_wrapper: RunContextWrapper[Any] = RunContextWrapper(context=None)
|
|
# Execute the computer action.
|
|
output_item = await ComputerAction.execute(
|
|
agent=agent,
|
|
action=tool_run,
|
|
hooks=run_hooks,
|
|
context_wrapper=context_wrapper,
|
|
config=RunConfig(),
|
|
)
|
|
# Both global and per-agent hooks should have been called once.
|
|
assert len(run_hooks.started) == 1 and len(agent_hooks.started) == 1
|
|
assert len(run_hooks.ended) == 1 and len(agent_hooks.ended) == 1
|
|
# The hook invocations should refer to our agent and tool.
|
|
assert run_hooks.started[0][0] is agent
|
|
assert run_hooks.ended[0][0] is agent
|
|
assert run_hooks.started[0][1] is comptool
|
|
assert run_hooks.ended[0][1] is comptool
|
|
# The result passed to on_tool_end should be the raw screenshot string.
|
|
assert run_hooks.ended[0][2] == "xyz"
|
|
assert agent_hooks.ended[0][2] == "xyz"
|
|
# The computer should have performed a click then a screenshot.
|
|
assert computer.calls == [("click", (1, 2, "left")), ("screenshot", ())]
|
|
# The returned item should include the agent, output string, and a ComputerCallOutput.
|
|
assert output_item.agent is agent
|
|
assert isinstance(output_item, ToolCallOutputItem)
|
|
assert output_item.output == "data:image/png;base64,xyz"
|
|
raw = cast(dict[str, Any], output_item.raw_item)
|
|
# Raw item is a dict-like mapping with expected output fields.
|
|
assert raw["type"] == "computer_call_output"
|
|
assert raw["output"]["type"] == "computer_screenshot"
|
|
assert "image_url" in raw["output"]
|
|
assert raw["output"]["image_url"].endswith("xyz")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_emits_function_span() -> None:
|
|
computer = LoggingComputer(screenshot_return="trace_img")
|
|
comptool = ComputerTool(computer=computer)
|
|
tool_call = ResponseComputerToolCall(
|
|
id="tool_trace",
|
|
type="computer_call",
|
|
action=ActionScreenshot(type="screenshot"),
|
|
call_id="tool_trace",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
tool_run = ToolRunComputerAction(tool_call=tool_call, computer_tool=comptool)
|
|
agent = Agent(name="test_agent_trace", tools=[comptool])
|
|
|
|
set_tracing_disabled(False)
|
|
with trace("computer-span-test"):
|
|
result = await ComputerAction.execute(
|
|
agent=agent,
|
|
action=tool_run,
|
|
hooks=RunHooks[Any](),
|
|
context_wrapper=RunContextWrapper(context=None),
|
|
config=RunConfig(),
|
|
)
|
|
|
|
assert isinstance(result, ToolCallOutputItem)
|
|
assert ComputerAction.TRACE_TOOL_NAME == "computer"
|
|
function_span = _get_function_span(ComputerAction.TRACE_TOOL_NAME)
|
|
span_data = cast(dict[str, Any], function_span["span_data"])
|
|
assert span_data.get("input") is not None
|
|
assert cast(str, span_data.get("output", "")).startswith("data:image/png;base64,")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runner_trace_lists_ga_computer_tool_name() -> None:
|
|
SPAN_PROCESSOR_TESTING.clear()
|
|
|
|
computer = LoggingComputer(screenshot_return="trace_img")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="tool_trace_agent_tools",
|
|
type="computer_call",
|
|
action=ActionScreenshot(type="screenshot"),
|
|
call_id="tool_trace_agent_tools",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
model = FakeModel(tracing_enabled=True)
|
|
model.add_multiple_turn_outputs(
|
|
[
|
|
[tool_call],
|
|
[get_text_message("done")],
|
|
]
|
|
)
|
|
agent = Agent(
|
|
name="test_agent_trace_tools",
|
|
model=model,
|
|
tools=[ComputerTool(computer=computer)],
|
|
)
|
|
|
|
set_tracing_disabled(False)
|
|
with trace("computer-agent-span-test"):
|
|
result = await Runner.run(agent, input="take a screenshot")
|
|
|
|
assert result.final_output == "done"
|
|
agent_span = _get_agent_span(agent.name)
|
|
span_data = cast(dict[str, Any], agent_span["span_data"])
|
|
assert span_data["tools"] == ["computer"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_emits_batched_actions_in_function_span() -> None:
|
|
computer = LoggingComputer(screenshot_return="trace_img")
|
|
comptool = ComputerTool(computer=computer)
|
|
tool_call = ResponseComputerToolCall(
|
|
id="tool_trace_batch",
|
|
type="computer_call",
|
|
actions=[
|
|
BatchedClick(type="click", x=5, y=6, button="left"),
|
|
BatchedType(type="type", text="batched"),
|
|
],
|
|
call_id="tool_trace_batch",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
tool_run = ToolRunComputerAction(tool_call=tool_call, computer_tool=comptool)
|
|
agent = Agent(name="test_agent_trace_batch", tools=[comptool])
|
|
|
|
set_tracing_disabled(False)
|
|
with trace("computer-batch-span-test"):
|
|
result = await ComputerAction.execute(
|
|
agent=agent,
|
|
action=tool_run,
|
|
hooks=RunHooks[Any](),
|
|
context_wrapper=RunContextWrapper(context=None),
|
|
config=RunConfig(),
|
|
)
|
|
|
|
assert isinstance(result, ToolCallOutputItem)
|
|
function_span = _get_function_span(ComputerAction.TRACE_TOOL_NAME)
|
|
span_data = cast(dict[str, Any], function_span["span_data"])
|
|
assert json.loads(cast(str, span_data["input"])) == [
|
|
{"type": "click", "x": 5, "y": 6, "button": "left"},
|
|
{"type": "type", "text": "batched"},
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_execute_redacts_span_error_when_sensitive_data_disabled() -> None:
|
|
secret_error = "computer secret output"
|
|
|
|
class FailingComputer(LoggingComputer):
|
|
def screenshot(self) -> str:
|
|
raise RuntimeError(secret_error)
|
|
|
|
computer = FailingComputer()
|
|
comptool = ComputerTool(computer=computer)
|
|
tool_call = ResponseComputerToolCall(
|
|
id="tool_trace_error",
|
|
type="computer_call",
|
|
action=ActionScreenshot(type="screenshot"),
|
|
call_id="tool_trace_error",
|
|
pending_safety_checks=[],
|
|
status="completed",
|
|
)
|
|
tool_run = ToolRunComputerAction(tool_call=tool_call, computer_tool=comptool)
|
|
agent = Agent(name="test_agent_trace_error", tools=[comptool])
|
|
|
|
set_tracing_disabled(False)
|
|
with trace("computer-span-redaction-test"):
|
|
with pytest.raises(RuntimeError, match=secret_error):
|
|
await ComputerAction.execute(
|
|
agent=agent,
|
|
action=tool_run,
|
|
hooks=RunHooks[Any](),
|
|
context_wrapper=RunContextWrapper(context=None),
|
|
config=RunConfig(trace_include_sensitive_data=False),
|
|
)
|
|
|
|
function_span = _get_function_span(ComputerAction.TRACE_TOOL_NAME)
|
|
assert function_span.get("error") == {
|
|
"message": "Error running tool",
|
|
"data": {
|
|
"tool_name": ComputerAction.TRACE_TOOL_NAME,
|
|
"error": "Tool execution failed. Error details are redacted.",
|
|
},
|
|
}
|
|
assert secret_error not in json.dumps(function_span)
|
|
span_data = cast(dict[str, Any], function_span["span_data"])
|
|
assert span_data.get("input") is None
|
|
assert span_data.get("output") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_safety_check_acknowledged() -> None:
|
|
"""Safety checks should be acknowledged via the callback."""
|
|
|
|
computer = LoggingComputer(screenshot_return="img")
|
|
called: list[ComputerToolSafetyCheckData] = []
|
|
|
|
def on_sc(data: ComputerToolSafetyCheckData) -> bool:
|
|
called.append(data)
|
|
return True
|
|
|
|
tool = ComputerTool(computer=computer, on_safety_check=on_sc)
|
|
safety = PendingSafetyCheck(id="sc", code="c", message="m")
|
|
tool_call = ResponseComputerToolCall(
|
|
id="t1",
|
|
type="computer_call",
|
|
action=ActionClick(type="click", x=1, y=1, button="left"),
|
|
call_id="t1",
|
|
pending_safety_checks=[safety],
|
|
status="completed",
|
|
)
|
|
run_action = ToolRunComputerAction(tool_call=tool_call, computer_tool=tool)
|
|
agent = Agent(name="a", tools=[tool])
|
|
ctx = RunContextWrapper(context=None)
|
|
|
|
results = await run_loop.execute_computer_actions(
|
|
public_agent=agent,
|
|
actions=[run_action],
|
|
hooks=RunHooks[Any](),
|
|
context_wrapper=ctx,
|
|
config=RunConfig(),
|
|
)
|
|
|
|
assert len(results) == 1
|
|
raw = results[0].raw_item
|
|
assert isinstance(raw, dict)
|
|
assert raw.get("acknowledged_safety_checks") == [{"id": "sc", "code": "c", "message": "m"}]
|
|
assert len(called) == 1
|
|
assert called[0].safety_check.id == "sc"
|