wehub-resource-sync
636 lines
23 KiB
Python
636 lines
23 KiB
Python
"""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
|