"""Tests for ToolOutputTrimmer — the built-in call_model_input_filter for trimming large tool outputs from older conversation turns. """ from __future__ import annotations import copy import json from typing import Any, cast from unittest.mock import MagicMock import pytest from agents.extensions.tool_output_trimmer import ToolOutputTrimmer from agents.run_config import CallModelData, ModelInputData # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _user(text: str = "hello") -> dict[str, Any]: return {"role": "user", "content": text} def _assistant(text: str = "response") -> dict[str, Any]: return {"role": "assistant", "content": text} def _func_call(call_id: str, name: str, *, namespace: str | None = None) -> dict[str, Any]: item = {"type": "function_call", "call_id": call_id, "name": name, "arguments": "{}"} if namespace is not None: item["namespace"] = namespace return item def _func_output(call_id: str, output: str) -> dict[str, Any]: return {"type": "function_call_output", "call_id": call_id, "output": output} def _make_data(items: list[Any]) -> CallModelData[Any]: model_data = ModelInputData(input=items, instructions="You are helpful.") return CallModelData(model_data=model_data, agent=MagicMock(), context=None) def _output(result: ModelInputData, idx: int) -> Any: """Extract the ``output`` field from a result item (untyped for test convenience).""" item: Any = result.input[idx] return item["output"] # --------------------------------------------------------------------------- # Defaults # --------------------------------------------------------------------------- class TestDefaults: def test_default_values(self) -> None: trimmer = ToolOutputTrimmer() assert trimmer.recent_turns == 2 assert trimmer.max_output_chars == 500 assert trimmer.preview_chars == 200 assert trimmer.trimmable_tools is None def test_trimmable_tools_coerced_to_frozenset(self) -> None: trimmer = ToolOutputTrimmer(trimmable_tools=frozenset({"a", "b"})) assert isinstance(trimmer.trimmable_tools, frozenset) assert trimmer.trimmable_tools == frozenset({"a", "b"}) def test_trimmable_tools_from_list(self) -> None: trimmer = ToolOutputTrimmer(trimmable_tools=["search", "run_code"]) assert isinstance(trimmer.trimmable_tools, frozenset) assert "search" in trimmer.trimmable_tools assert "run_code" in trimmer.trimmable_tools def test_trimmable_tools_from_string(self) -> None: trimmer = ToolOutputTrimmer(trimmable_tools="search") assert isinstance(trimmer.trimmable_tools, frozenset) assert trimmer.trimmable_tools == frozenset({"search"}) # --------------------------------------------------------------------------- # Input validation # --------------------------------------------------------------------------- class TestValidation: def test_recent_turns_zero_raises(self) -> None: with pytest.raises(ValueError, match="recent_turns must be >= 1"): ToolOutputTrimmer(recent_turns=0) def test_recent_turns_negative_raises(self) -> None: with pytest.raises(ValueError, match="recent_turns must be >= 1"): ToolOutputTrimmer(recent_turns=-1) def test_max_output_chars_zero_raises(self) -> None: with pytest.raises(ValueError, match="max_output_chars must be >= 1"): ToolOutputTrimmer(max_output_chars=0) def test_preview_chars_negative_raises(self) -> None: with pytest.raises(ValueError, match="preview_chars must be >= 0"): ToolOutputTrimmer(preview_chars=-1) def test_preview_chars_zero_allowed(self) -> None: trimmer = ToolOutputTrimmer(preview_chars=0) assert trimmer.preview_chars == 0 def test_trimmable_tools_bytes_raises(self) -> None: with pytest.raises(ValueError, match="trimmable_tools must be a string or iterable"): ToolOutputTrimmer(trimmable_tools=b"search") # type: ignore[arg-type] # --------------------------------------------------------------------------- # Boundary detection # --------------------------------------------------------------------------- class TestRecentBoundary: def test_empty_items(self) -> None: trimmer = ToolOutputTrimmer() assert trimmer._find_recent_boundary([]) == 0 def test_single_user_message(self) -> None: trimmer = ToolOutputTrimmer() assert trimmer._find_recent_boundary([_user()]) == 0 def test_two_user_messages_boundary_at_first(self) -> None: items = [_user("q1"), _assistant("a1"), _user("q2"), _assistant("a2")] trimmer = ToolOutputTrimmer(recent_turns=2) assert trimmer._find_recent_boundary(items) == 0 def test_three_user_messages(self) -> None: items = [ _user("q1"), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(recent_turns=2) assert trimmer._find_recent_boundary(items) == 2 def test_custom_recent_turns(self) -> None: items = [ _user("q1"), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), _user("q4"), _assistant("a4"), ] trimmer = ToolOutputTrimmer(recent_turns=3) # q4 at 6 (count=1), q3 at 4 (count=2), q2 at 2 (count=3) -> boundary=2 assert trimmer._find_recent_boundary(items) == 2 # --------------------------------------------------------------------------- # Trimming behavior # --------------------------------------------------------------------------- class TestTrimming: def test_empty_input(self) -> None: trimmer = ToolOutputTrimmer() data = _make_data([]) result = trimmer(data) assert result.input == [] def test_no_trimming_when_all_recent(self) -> None: """With only 1 user message, everything is recent.""" large = "x" * 1000 items = [ _user("q"), _func_call("c1", "search"), _func_output("c1", large), _assistant("a"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) assert _output(result, 2) == large def test_trims_large_old_output(self) -> None: """Large output in an old turn should be trimmed.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) trimmed = _output(result, 2) assert "[Trimmed:" in trimmed assert "search" in trimmed assert "1000 chars" in trimmed assert len(trimmed) < len(large) def test_preserves_small_old_output(self) -> None: """Small outputs should never be trimmed.""" small = "x" * 100 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", small), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(max_output_chars=500) result = trimmer(_make_data(items)) assert _output(result, 2) == small def test_respects_trimmable_tools_allowlist(self) -> None: """Only outputs from tools in trimmable_tools should be trimmed.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large), _func_call("c2", "resolve_entity"), _func_output("c2", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools=frozenset({"search"})) result = trimmer(_make_data(items)) # search output trimmed assert "[Trimmed:" in _output(result, 2) # resolve_entity output preserved assert _output(result, 4) == large def test_string_trimmable_tools_allowlist_matches_single_tool_name(self) -> None: """A string trimmable_tools value should match one tool name, not characters.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large), _func_call("c2", "s"), _func_output("c2", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools="search") result = trimmer(_make_data(items)) assert "[Trimmed:" in _output(result, 2) assert _output(result, 4) == large def test_respects_qualified_tool_names_allowlist(self) -> None: """Qualified allowlist entries should match namespaced function tools.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "lookup_account", namespace="billing"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools=frozenset({"billing.lookup_account"})) result = trimmer(_make_data(items)) assert "[Trimmed:" in _output(result, 2) assert "billing.lookup_account" in _output(result, 2) def test_namespaced_tools_still_match_bare_allowlist_entries(self) -> None: """Bare allowlist entries remain valid for namespaced tools.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "lookup_account", namespace="billing"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools=frozenset({"lookup_account"})) result = trimmer(_make_data(items)) assert "[Trimmed:" in _output(result, 2) assert "billing.lookup_account" in _output(result, 2) def test_synthetic_same_name_namespace_uses_bare_display_name(self) -> None: """Deferred synthetic namespaces should not display as `name.name`.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "get_weather", namespace="get_weather"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools=frozenset({"get_weather"})) result = trimmer(_make_data(items)) assert "[Trimmed:" in _output(result, 2) assert "get_weather.get_weather" not in _output(result, 2) assert "get_weather" in _output(result, 2) def test_trims_tool_search_output_tool_definitions(self) -> None: """Large tool_search_output tool definitions should be structurally trimmed.""" verbose_schema = { "type": "object", "description": "schema " * 200, "properties": { "customer_id": { "type": "string", "description": "customer id " * 200, "default": "cust_123", } }, "required": ["customer_id"], } items = [ _user("q1"), {"type": "tool_search_call", "call_id": "ts1", "arguments": {"query": "profile"}}, { "type": "tool_search_output", "call_id": "ts1", "tools": [ { "type": "function", "name": "lookup_account", "description": "tool description " * 200, "parameters": verbose_schema, } ], }, _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] original_len = len(json.dumps(items[2]["tools"], sort_keys=True)) trimmer = ToolOutputTrimmer(max_output_chars=400, preview_chars=60) result = trimmer(_make_data(items)) trimmed_item_dict = cast(dict[str, Any], result.input[2]) assert trimmed_item_dict["type"] == "tool_search_output" trimmed_tools = list(trimmed_item_dict["tools"]) assert trimmed_tools[0]["name"] == "lookup_account" assert "description" not in trimmed_tools[0]["parameters"] assert trimmed_tools[0]["parameters"]["properties"]["customer_id"]["default"] == "cust_123" assert len(json.dumps(trimmed_tools, sort_keys=True)) < original_len def test_trims_legacy_tool_search_output_results(self) -> None: """Legacy tool_search_output snapshots with free-text results should still trim.""" large = "x" * 2000 items = [ _user("q1"), {"type": "tool_search_call", "call_id": "ts1", "arguments": {"query": "profile"}}, { "type": "tool_search_output", "call_id": "ts1", "results": [{"text": large}], }, _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(max_output_chars=400, preview_chars=80) result = trimmer(_make_data(items)) trimmed_item = cast(dict[str, Any], result.input[2]) assert trimmed_item["type"] == "tool_search_output" assert "[Trimmed: tool_search output" in trimmed_item["results"][0]["text"] def test_trims_all_tools_when_allowlist_is_none(self) -> None: """When trimmable_tools is None, all tools are eligible.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "any_tool"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools=None) result = trimmer(_make_data(items)) assert "[Trimmed:" in _output(result, 2) def test_preserves_recent_large_output(self) -> None: """Large outputs in recent turns should never be trimmed.""" large = "x" * 1000 items = [ _user("q1"), _assistant("a1"), _user("q2"), _func_call("c1", "search"), _func_output("c1", large), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) assert _output(result, 4) == large def test_does_not_mutate_original_items(self) -> None: """The filter must not mutate the original input items.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] original = copy.deepcopy(items) trimmer = ToolOutputTrimmer() trimmer(_make_data(items)) assert items == original def test_preserves_instructions(self) -> None: """The instructions field should pass through unchanged.""" items: list[Any] = [_user("hi")] model_data = ModelInputData(input=items, instructions="Custom prompt") data: CallModelData[Any] = CallModelData( model_data=model_data, agent=MagicMock(), context=None ) trimmer = ToolOutputTrimmer() result = trimmer(data) assert result.instructions == "Custom prompt" def test_multiple_old_outputs_trimmed(self) -> None: """Multiple large outputs in old turns should all be trimmed.""" large1 = "a" * 1000 large2 = "b" * 2000 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large1), _func_call("c2", "execute"), _func_output("c2", large2), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) assert "[Trimmed:" in _output(result, 2) assert "[Trimmed:" in _output(result, 4) assert "search" in _output(result, 2) assert "execute" in _output(result, 4) def test_custom_preview_chars(self) -> None: """Preview length should respect the preview_chars setting.""" large = "abcdefghij" * 100 # 1000 chars items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(preview_chars=50) result = trimmer(_make_data(items)) trimmed = _output(result, 2) # The preview portion should be exactly 50 chars of the original assert "abcdefghij" * 5 in trimmed def test_preserves_user_and_assistant_messages(self) -> None: """User and assistant messages are never modified.""" items = [ _user("important"), _assistant("detailed " * 100), _user("follow up"), _assistant("another"), _user("final"), _assistant("done"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) assert result.input == items # --------------------------------------------------------------------------- # Sliding window behavior # --------------------------------------------------------------------------- class TestSlidingWindow: """Verify the trimmer acts as a sliding window across turns.""" def test_turn3_trims_turn1(self) -> None: """On turn 3, turn 1 outputs should be trimmed.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", large), _assistant("a1"), _user("q2"), _func_call("c2", "search"), _func_output("c2", large), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) # Turn 1 (old) trimmed assert "[Trimmed:" in _output(result, 2) # Turn 2 (recent) preserved assert _output(result, 6) == large def test_turn4_trims_turns_1_and_2(self) -> None: """On turn 4, turns 1 and 2 outputs should both be trimmed.""" large = "x" * 1000 items = [ _user("q1"), _func_call("c1", "s"), _func_output("c1", large), _assistant("a1"), _user("q2"), _func_call("c2", "s"), _func_output("c2", large), _assistant("a2"), _user("q3"), _func_call("c3", "s"), _func_output("c3", large), _assistant("a3"), _user("q4"), _assistant("a4"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) # Turns 1 and 2 trimmed assert "[Trimmed:" in _output(result, 2) assert "[Trimmed:" in _output(result, 6) # Turn 3 (recent) preserved assert _output(result, 10) == large # --------------------------------------------------------------------------- # Edge cases # --------------------------------------------------------------------------- class TestEdgeCases: def test_skips_trim_when_summary_would_exceed_original(self) -> None: """When preview_chars is large relative to the output, the summary can be longer than the original. In that case the output should be left untouched.""" # Output is 501 chars (just above default max_output_chars=500). # With preview_chars=490, the summary header + 490-char preview + "..." will # easily exceed 501 chars, so trimming should be skipped. borderline = "x" * 501 items = [ _user("q1"), _func_call("c1", "search"), _func_output("c1", borderline), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(max_output_chars=500, preview_chars=490) result = trimmer(_make_data(items)) # Output left untouched because summary would be longer assert _output(result, 2) == borderline def test_unknown_tool_name_fallback(self) -> None: """When a function_call_output has no matching function_call, the summary should show 'unknown_tool' instead of a blank name.""" large = "x" * 1000 # Deliberately omit the _func_call so the call_id has no name mapping items = [ _user("q1"), _func_output("orphan_id", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer() result = trimmer(_make_data(items)) trimmed = _output(result, 1) assert "unknown_tool" in trimmed assert "[Trimmed:" in trimmed def test_unresolved_tool_skipped_with_allowlist(self) -> None: """When trimmable_tools is set and the tool name can't be resolved, the output should NOT be trimmed (empty string won't match the allowlist).""" large = "x" * 1000 items = [ _user("q1"), _func_output("orphan_id", large), _assistant("a1"), _user("q2"), _assistant("a2"), _user("q3"), _assistant("a3"), ] trimmer = ToolOutputTrimmer(trimmable_tools=frozenset({"search"})) result = trimmer(_make_data(items)) # Unresolved tool name is "" which is not in the allowlist — left untouched assert _output(result, 1) == large