"""Tests for the parser module.
Tests all parsing and analysis functions:
- compute_hash: Content hashing
- detect_waste_signals: Waste signal detection
- is_rag_content: RAG content detection
- parse_message_to_blocks: Single message parsing
- parse_messages: Multi-message parsing
- find_tool_units: Tool call/response pairing
- get_message_content_text: Content extraction
"""
from unittest.mock import Mock
import pytest
from headroom.parser import (
_coerce_tool_call_to_dict,
compute_hash,
detect_waste_signals,
find_tool_units,
get_message_content_text,
is_rag_content,
parse_message_to_blocks,
parse_messages,
)
# --- Streaming SDK tool-call objects (issue #1312) ---
class _FakeDeltaToolCallFunction:
"""Mimics openai.types...ChoiceDeltaToolCallFunction: attribute access,
no `.get()`."""
def __init__(self, name: str, arguments: str) -> None:
self.name = name
self.arguments = arguments
class _FakeChoiceDeltaToolCall:
"""Mimics the OpenAI SDK streaming tool-call object that the Agno
wrapper surfaces. It is a Pydantic-style model — attribute access only,
crucially with NO `.get()` — which is exactly what triggered issue
#1312 (`'ChoiceDeltaToolCall' object has no attribute 'get'`)."""
def __init__(self, id: str, name: str, arguments: str, index: int = 0) -> None:
self.id = id
self.index = index
self.type = "function"
self.function = _FakeDeltaToolCallFunction(name, arguments)
# --- Fixtures ---
@pytest.fixture
def mock_tokenizer():
"""Mock tokenizer that returns predictable token counts."""
tokenizer = Mock()
# Simple mock: 1 token per 4 characters
tokenizer.count_text = Mock(side_effect=lambda text: len(text) // 4 + 1)
return tokenizer
@pytest.fixture
def system_message():
"""Basic system message."""
return {"role": "system", "content": "You are a helpful assistant."}
@pytest.fixture
def user_message():
"""Basic user message."""
return {"role": "user", "content": "Hello, how are you?"}
@pytest.fixture
def assistant_message():
"""Basic assistant message."""
return {"role": "assistant", "content": "I'm doing well, thank you!"}
@pytest.fixture
def tool_call_message():
"""Assistant message with tool calls."""
return {
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {"name": "search_user", "arguments": '{"user_id": "12345"}'},
}
],
}
@pytest.fixture
def tool_result_message():
"""Tool result message."""
return {
"role": "tool",
"tool_call_id": "call_abc123",
"content": '{"id": "12345", "name": "Alice", "email": "alice@example.com"}',
}
@pytest.fixture
def multimodal_message():
"""User message with multimodal content (list format)."""
return {
"role": "user",
"content": [
{"type": "text", "text": "Analyze this image:"},
{"type": "image", "source": {"type": "base64", "data": "..."}},
{"type": "text", "text": "What do you see?"},
],
}
@pytest.fixture
def rag_user_message():
"""User message containing RAG content markers."""
return {
"role": "user",
"content": "[Document 1] Here is the relevant context from our knowledge base. [Source: docs/manual.md]",
}
@pytest.fixture
def html_waste_text():
"""Text containing HTML noise."""
return ""
@pytest.fixture
def base64_waste_text():
"""Text containing base64 encoded data."""
return "Data: " + "A" * 60 + "=="
@pytest.fixture
def whitespace_waste_text():
"""Text with excessive whitespace."""
return "Line 1\n\n\n\nLine 2 extra spaces"
@pytest.fixture
def json_bloat_text():
"""Text containing large JSON block (>500 chars).
Uses spaces and punctuation to avoid base64 pattern matching.
"""
# Use content that won't match base64 pattern (needs non-base64 chars)
content = "This is a long text value. " * 25 # ~675 chars
return '{"data": "' + content + '"}'
# --- TestComputeHash ---
class TestComputeHash:
"""Tests for compute_hash function."""
def test_consistent_hash(self):
"""Same text produces same hash."""
text = "Hello, world!"
hash1 = compute_hash(text)
hash2 = compute_hash(text)
assert hash1 == hash2
def test_different_texts_different_hashes(self):
"""Different texts produce different hashes."""
hash1 = compute_hash("Hello")
hash2 = compute_hash("World")
assert hash1 != hash2
def test_hash_length_16(self):
"""Hash is truncated to 16 characters."""
text = "Any text content"
hash_result = compute_hash(text)
assert len(hash_result) == 16
def test_empty_string_hash(self):
"""Empty string produces valid hash."""
hash_result = compute_hash("")
assert len(hash_result) == 16
assert hash_result.isalnum()
def test_unicode_text_hash(self):
"""Unicode text produces valid hash."""
hash_result = compute_hash("Hello \\u4e16\\u754c")
assert len(hash_result) == 16
# --- TestDetectWasteSignals ---
class TestDetectWasteSignals:
"""Tests for detect_waste_signals function."""
def test_detect_html_tags(self, mock_tokenizer, html_waste_text):
"""Detects HTML tags as waste."""
signals = detect_waste_signals(html_waste_text, mock_tokenizer)
assert signals.html_noise_tokens > 0
def test_detect_html_comments(self, mock_tokenizer):
"""Detects HTML comments as waste."""
text = "Some text more text"
signals = detect_waste_signals(text, mock_tokenizer)
assert signals.html_noise_tokens > 0
def test_detect_base64(self, mock_tokenizer, base64_waste_text):
"""Detects base64 encoded content as waste."""
signals = detect_waste_signals(base64_waste_text, mock_tokenizer)
assert signals.base64_tokens > 0
def test_detect_excessive_whitespace(self, mock_tokenizer, whitespace_waste_text):
"""Detects excessive whitespace as waste."""
signals = detect_waste_signals(whitespace_waste_text, mock_tokenizer)
assert signals.whitespace_tokens >= 0 # May be 0 if normalized tokens <= matches
def test_detect_json_bloat(self, mock_tokenizer, json_bloat_text):
"""Detects large JSON blocks as bloat."""
# Need to ensure the mock returns >500 tokens for JSON bloat
# The JSON pattern requires the matched block to have >500 tokens
mock_tokenizer.count_text = Mock(side_effect=lambda text: len(text))
signals = detect_waste_signals(json_bloat_text, mock_tokenizer)
assert signals.json_bloat_tokens > 0
def test_empty_text_no_waste(self, mock_tokenizer):
"""Empty text returns zero waste signals."""
signals = detect_waste_signals("", mock_tokenizer)
assert signals.total() == 0
def test_combined_waste_signals(self, mock_tokenizer):
"""Multiple waste types are detected together."""
text = "Hello
" + "B" * 60 + "== and "
signals = detect_waste_signals(text, mock_tokenizer)
assert signals.html_noise_tokens > 0
assert signals.base64_tokens > 0
def test_clean_text_no_waste(self, mock_tokenizer):
"""Clean text produces minimal waste signals."""
text = "This is a normal sentence without any waste."
signals = detect_waste_signals(text, mock_tokenizer)
assert signals.html_noise_tokens == 0
assert signals.base64_tokens == 0
assert signals.json_bloat_tokens == 0
# --- TestIsRagContent ---
class TestIsRagContent:
"""Tests for is_rag_content function."""
def test_document_markers(self):
"""Detects [Document N] markers."""
text = "[Document 1] This is the first document. [Document 2] Second document."
assert is_rag_content(text) is True
def test_source_markers(self):
"""Detects [Source: ...] markers."""
text = "[Source: knowledge_base/docs.md] Here is the information."
assert is_rag_content(text) is True
def test_context_tags(self):
"""Detects and tags."""
assert is_rag_content("Retrieved content here") is True
assert is_rag_content("Document content") is True
def test_retrieved_from_marker(self):
"""Detects 'Retrieved from:' marker."""
text = "Retrieved from: https://example.com/docs\nHere is the content."
assert is_rag_content(text) is True
def test_knowledge_base_marker(self):
"""Detects 'From the knowledge base:' marker."""
text = "From the knowledge base: This is relevant information."
assert is_rag_content(text) is True
def test_not_rag_content(self):
"""Regular text is not detected as RAG content."""
text = "Hello, how can I help you today?"
assert is_rag_content(text) is False
def test_case_insensitive(self):
"""RAG detection is case insensitive."""
assert is_rag_content("[DOCUMENT 1] Content") is True
assert is_rag_content("retrieved FROM: somewhere") is True
# --- TestParseMessageToBlocks ---
class TestParseMessageToBlocks:
"""Tests for parse_message_to_blocks function."""
def test_system_message_block(self, mock_tokenizer, system_message):
"""System message creates system block."""
blocks = parse_message_to_blocks(system_message, 0, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "system"
assert blocks[0].text == "You are a helpful assistant."
assert blocks[0].source_index == 0
def test_user_message_block(self, mock_tokenizer, user_message):
"""User message creates user block."""
blocks = parse_message_to_blocks(user_message, 1, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "user"
assert blocks[0].text == "Hello, how are you?"
assert blocks[0].source_index == 1
def test_assistant_message_block(self, mock_tokenizer, assistant_message):
"""Assistant message creates assistant block."""
blocks = parse_message_to_blocks(assistant_message, 2, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "assistant"
assert blocks[0].text == "I'm doing well, thank you!"
def test_tool_result_block(self, mock_tokenizer, tool_result_message):
"""Tool result creates tool_result block with tool_call_id."""
blocks = parse_message_to_blocks(tool_result_message, 3, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "tool_result"
assert blocks[0].flags.get("tool_call_id") == "call_abc123"
def test_rag_detection_in_user_message(self, mock_tokenizer, rag_user_message):
"""User message with RAG markers creates rag block."""
blocks = parse_message_to_blocks(rag_user_message, 0, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "rag"
def test_multimodal_content(self, mock_tokenizer, multimodal_message):
"""Multimodal content (list with text parts) is extracted."""
blocks = parse_message_to_blocks(multimodal_message, 0, mock_tokenizer)
assert len(blocks) == 1
assert "Analyze this image:" in blocks[0].text
assert "What do you see?" in blocks[0].text
def test_tool_calls_create_separate_blocks(self, mock_tokenizer, tool_call_message):
"""Tool calls create separate tool_call blocks."""
blocks = parse_message_to_blocks(tool_call_message, 0, mock_tokenizer)
# Should have tool_call blocks (no content block since content is None)
tool_call_blocks = [b for b in blocks if b.kind == "tool_call"]
assert len(tool_call_blocks) == 1
assert tool_call_blocks[0].flags.get("tool_call_id") == "call_abc123"
assert tool_call_blocks[0].flags.get("function_name") == "search_user"
assert "search_user" in tool_call_blocks[0].text
def test_empty_message_creates_block(self, mock_tokenizer):
"""Empty message (no content or tool_calls) creates minimal block."""
empty_msg = {"role": "assistant"}
blocks = parse_message_to_blocks(empty_msg, 0, mock_tokenizer)
assert len(blocks) == 1
assert blocks[0].kind == "unknown"
assert blocks[0].text == ""
def test_message_with_content_and_tool_calls(self, mock_tokenizer):
"""Message with both content and tool_calls creates multiple blocks."""
msg = {
"role": "assistant",
"content": "Let me search for that.",
"tool_calls": [{"id": "call_xyz", "function": {"name": "search", "arguments": "{}"}}],
}
blocks = parse_message_to_blocks(msg, 0, mock_tokenizer)
kinds = [b.kind for b in blocks]
assert "assistant" in kinds
assert "tool_call" in kinds
def test_waste_signals_in_flags(self, mock_tokenizer, html_waste_text):
"""Waste signals are added to block flags."""
msg = {"role": "user", "content": html_waste_text}
blocks = parse_message_to_blocks(msg, 0, mock_tokenizer)
assert "waste_signals" in blocks[0].flags
assert blocks[0].flags["waste_signals"]["html_noise"] > 0
def test_content_hash_generated(self, mock_tokenizer, user_message):
"""Content hash is generated for blocks."""
blocks = parse_message_to_blocks(user_message, 0, mock_tokenizer)
assert len(blocks[0].content_hash) == 16
def test_tokens_estimated(self, mock_tokenizer, user_message):
"""Token count is estimated."""
blocks = parse_message_to_blocks(user_message, 0, mock_tokenizer)
assert blocks[0].tokens_est > 0
class TestStreamingToolCallObjects:
"""Regression coverage for issue #1312: streaming integrations (Agno
over OpenAILike) can hand the parser raw OpenAI SDK `ChoiceDeltaToolCall`
objects instead of OpenAI-format dicts. The parser called `.get()` on
them and crashed the whole agent run with
`'ChoiceDeltaToolCall' object has no attribute 'get'`. Both the parser
call sites must now tolerate attribute-style tool-call objects."""
def test_coerce_dict_is_passthrough_identity(self):
d = {"id": "call_1", "function": {"name": "f", "arguments": "{}"}}
# A dict must be returned untouched (same object) — no needless copy.
assert _coerce_tool_call_to_dict(d) is d
def test_coerce_sdk_object_flattens_to_openai_dict(self):
tc = _FakeChoiceDeltaToolCall("call_1", "search", '{"q": "x"}')
out = _coerce_tool_call_to_dict(tc)
assert out == {
"id": "call_1",
"type": "function",
"function": {"name": "search", "arguments": '{"q": "x"}'},
}
def test_coerce_object_with_dict_function(self):
# Some providers nest a dict `function` on an attribute-style object.
class _TC:
id = "call_2"
type = "function"
function = {"name": "g", "arguments": "1"}
out = _coerce_tool_call_to_dict(_TC())
assert out["function"] == {"name": "g", "arguments": "1"}
def test_coerce_none_degrades_to_empty_dict(self):
assert _coerce_tool_call_to_dict(None) == {}
def test_parse_message_to_blocks_with_sdk_tool_call(self, mock_tokenizer):
"""The original crash site: parsing an assistant message whose
tool_calls are SDK objects must produce a tool_call block, not
raise AttributeError."""
tc = _FakeChoiceDeltaToolCall("call_abc", "dummy_tool", '{"query": "test"}')
msg = {"role": "assistant", "content": "", "tool_calls": [tc]}
blocks = parse_message_to_blocks(msg, 0, mock_tokenizer)
tool_call_blocks = [b for b in blocks if b.kind == "tool_call"]
assert len(tool_call_blocks) == 1
assert tool_call_blocks[0].flags.get("tool_call_id") == "call_abc"
assert tool_call_blocks[0].flags.get("function_name") == "dummy_tool"
assert "dummy_tool" in tool_call_blocks[0].text
def test_find_tool_units_with_sdk_tool_call(self):
"""The second `.get()` site: find_tool_units must still pair an
SDK-object tool_call with its tool response message."""
tc = _FakeChoiceDeltaToolCall("call_abc", "dummy_tool", "{}")
messages = [
{"role": "assistant", "content": "", "tool_calls": [tc]},
{"role": "tool", "content": "result", "tool_call_id": "call_abc"},
]
units = find_tool_units(messages)
assert units == [(0, [1])]
# --- TestParseMessages ---
class TestParseMessages:
"""Tests for parse_messages function."""
def test_parse_all_messages(self, mock_tokenizer, sample_messages):
"""All messages are parsed into blocks."""
blocks, breakdown, waste = parse_messages(sample_messages, mock_tokenizer)
assert len(blocks) >= len(sample_messages)
def test_block_breakdown(self, mock_tokenizer, sample_messages):
"""Block breakdown counts tokens per kind."""
blocks, breakdown, waste = parse_messages(sample_messages, mock_tokenizer)
assert "system" in breakdown
assert "user" in breakdown
assert "assistant" in breakdown
assert all(v > 0 for v in breakdown.values())
def test_waste_signals_accumulated(self, mock_tokenizer):
"""Waste signals are accumulated across messages."""
messages = [
{"role": "user", "content": "HTML here
"},
{"role": "assistant", "content": "More HTML"},
]
blocks, breakdown, waste = parse_messages(messages, mock_tokenizer)
assert waste.html_noise_tokens > 0
def test_empty_messages(self, mock_tokenizer):
"""Empty message list returns empty results."""
blocks, breakdown, waste = parse_messages([], mock_tokenizer)
assert blocks == []
assert breakdown == {}
assert waste.total() == 0
def test_multiple_tool_calls_parsed(self, mock_tokenizer, sample_messages_with_tools):
"""Messages with tool calls are parsed correctly."""
blocks, breakdown, waste = parse_messages(sample_messages_with_tools, mock_tokenizer)
tool_call_blocks = [b for b in blocks if b.kind == "tool_call"]
tool_result_blocks = [b for b in blocks if b.kind == "tool_result"]
assert len(tool_call_blocks) >= 1
assert len(tool_result_blocks) >= 1
# --- TestRereadDetection ---
class TestRereadDetection:
"""Tests for cross-message re-read detection in parse_messages."""
LARGE_CONTENT = "def handler(event):\n return process(event)\n" * 10 # > 200 chars
def _expected_tokens(self, text):
"""Mirror mock_tokenizer + message overhead used for tool_result blocks."""
return len(text) // 4 + 1 + 4
@staticmethod
def _filler(n):
"""Interleaved turns that push a repeat beyond the polling gap."""
return [
{"role": "assistant" if i % 2 == 0 else "user", "content": f"step {i} of the task"}
for i in range(n)
]
def test_reread_detected_openai_tool_messages(self, mock_tokenizer):
"""Identical large tool outputs far apart count as re-read."""
messages = (
[{"role": "tool", "tool_call_id": "c1", "content": self.LARGE_CONTENT}]
+ self._filler(4)
+ [{"role": "tool", "tool_call_id": "c2", "content": self.LARGE_CONTENT}]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
def test_reread_detected_anthropic_tool_result_blocks(self, mock_tokenizer):
"""Anthropic-format tool_result parts are matched by content, not id."""
part = {"type": "tool_result", "tool_use_id": "t1", "content": self.LARGE_CONTENT}
part2 = {"type": "tool_result", "tool_use_id": "t2", "content": self.LARGE_CONTENT}
messages = (
[{"role": "user", "content": [part]}]
+ self._filler(4)
+ [{"role": "user", "content": [part2]}]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
def test_three_occurrences_count_repeats_only(self, mock_tokenizer):
"""First serve is free; every distant repeat is counted."""
msg = {"role": "tool", "tool_call_id": "c", "content": self.LARGE_CONTENT}
messages = [dict(msg)] + self._filler(4) + [dict(msg)] + self._filler(4) + [dict(msg)]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 2 * self._expected_tokens(self.LARGE_CONTENT)
def test_single_occurrence_no_signal(self, mock_tokenizer):
"""One large tool result is not a re-read."""
messages = [{"role": "tool", "tool_call_id": "c1", "content": self.LARGE_CONTENT}]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_short_duplicates_ignored(self, mock_tokenizer):
"""Trivially short outputs (\"ok\") legitimately repeat and are skipped."""
messages = [
{"role": "tool", "tool_call_id": "c1", "content": "ok"},
{"role": "tool", "tool_call_id": "c2", "content": "ok"},
{"role": "tool", "tool_call_id": "c3", "content": "ok"},
]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_same_message_duplicates_ignored(self, mock_tokenizer):
"""Duplicates within a single message are not a re-read."""
part = {"type": "tool_result", "tool_use_id": "t1", "content": self.LARGE_CONTENT}
part2 = {"type": "tool_result", "tool_use_id": "t2", "content": self.LARGE_CONTENT}
messages = [{"role": "user", "content": [part, part2]}]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_mixed_same_message_duplicate_not_counted(self, mock_tokenizer):
"""A duplicate inside the original message stays excluded even when
a later message also re-serves the content."""
part = {"type": "tool_result", "tool_use_id": "t1", "content": self.LARGE_CONTENT}
part2 = {"type": "tool_result", "tool_use_id": "t2", "content": self.LARGE_CONTENT}
part3 = {"type": "tool_result", "tool_use_id": "t3", "content": self.LARGE_CONTENT}
messages = (
[{"role": "user", "content": [part, part2]}]
+ self._filler(4)
+ [{"role": "user", "content": [part3]}]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
def test_adjacent_polling_repeats_ignored(self, mock_tokenizer):
"""Back-to-back identical results (poll loop) are not re-reads."""
messages = [
{"role": "tool", "tool_call_id": "c1", "content": self.LARGE_CONTENT},
{"role": "assistant", "content": "Still pending, checking again."},
{"role": "tool", "tool_call_id": "c2", "content": self.LARGE_CONTENT},
]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_polling_chain_never_accumulates(self, mock_tokenizer):
"""Each poll advances the baseline — long chains stay at zero."""
msg = {"role": "tool", "tool_call_id": "c", "content": self.LARGE_CONTENT}
nudge = {"role": "assistant", "content": "polling"}
messages = [dict(msg), dict(nudge), dict(msg), dict(nudge), dict(msg), dict(nudge)]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_distant_repeat_after_polling_chain_counts(self, mock_tokenizer):
"""A far repeat counts even when earlier repeats were polling."""
msg = {"role": "tool", "tool_call_id": "c", "content": self.LARGE_CONTENT}
messages = (
[dict(msg), {"role": "assistant", "content": "polling"}, dict(msg)]
+ self._filler(4)
+ [dict(msg)]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
def test_reread_in_total_and_dict(self):
"""reread_tokens participates in total() and to_dict()."""
from headroom.config import WasteSignals
ws = WasteSignals(reread_tokens=42)
assert ws.total() == 42
assert ws.to_dict()["reread"] == 42
# --- TestFindToolUnits ---
class TestFindToolUnits:
"""Tests for find_tool_units function."""
def test_finds_tool_call_and_responses(self, sample_messages_with_tools):
"""Finds matching tool call and response pairs."""
units = find_tool_units(sample_messages_with_tools)
assert len(units) >= 1
# Each unit is (assistant_index, [tool_response_indices])
assistant_idx, response_indices = units[0]
assert response_indices # Should have at least one response
def test_multiple_tool_calls_same_assistant(self):
"""Multiple tool calls from same assistant are grouped."""
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Search both"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{"id": "call_1", "function": {"name": "search", "arguments": "{}"}},
{"id": "call_2", "function": {"name": "fetch", "arguments": "{}"}},
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "result 1"},
{"role": "tool", "tool_call_id": "call_2", "content": "result 2"},
]
units = find_tool_units(messages)
assert len(units) == 1
assistant_idx, response_indices = units[0]
assert len(response_indices) == 2
def test_no_tool_units(self):
"""Returns empty list when no tool calls present."""
messages = [
{"role": "system", "content": "Hello"},
{"role": "user", "content": "Hi"},
{"role": "assistant", "content": "Hello!"},
]
units = find_tool_units(messages)
assert units == []
def test_orphaned_tool_response(self):
"""Tool response without matching assistant is not included."""
messages = [
{"role": "system", "content": "Hello"},
{"role": "user", "content": "Hi"},
# Orphaned tool response - no assistant with tool_calls
{"role": "tool", "tool_call_id": "orphan_call", "content": "orphaned"},
{"role": "assistant", "content": "I don't have tools."},
]
units = find_tool_units(messages)
assert units == []
def test_tool_response_order_sorted(self):
"""Tool response indices are sorted."""
messages = [
{"role": "user", "content": "Do two things"},
{
"role": "assistant",
"tool_calls": [
{"id": "call_a", "function": {"name": "first", "arguments": "{}"}},
{"id": "call_b", "function": {"name": "second", "arguments": "{}"}},
],
},
{"role": "tool", "tool_call_id": "call_b", "content": "second result"},
{"role": "tool", "tool_call_id": "call_a", "content": "first result"},
]
units = find_tool_units(messages)
assert len(units) == 1
_, response_indices = units[0]
assert response_indices == sorted(response_indices)
def test_anthropic_format_tool_use_and_result(self):
"""Finds Anthropic format tool_use/tool_result pairs in content blocks."""
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "Take a screenshot"},
{
"role": "assistant",
"content": [
{"type": "text", "text": "Let me take a screenshot."},
{
"type": "tool_use",
"id": "toolu_123",
"name": "browser_screenshot",
"input": {},
},
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_123",
"content": "Screenshot taken successfully",
}
],
},
{"role": "user", "content": "Thanks!"},
]
units = find_tool_units(messages)
assert len(units) == 1
assistant_idx, response_indices = units[0]
assert assistant_idx == 2
assert response_indices == [3]
def test_anthropic_format_multiple_tool_uses(self):
"""Finds multiple Anthropic format tool_use blocks from same assistant."""
messages = [
{"role": "user", "content": "Do two things"},
{
"role": "assistant",
"content": [
{"type": "tool_use", "id": "toolu_a", "name": "first", "input": {}},
{"type": "tool_use", "id": "toolu_b", "name": "second", "input": {}},
],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_a", "content": "first done"},
{"type": "tool_result", "tool_use_id": "toolu_b", "content": "second done"},
],
},
]
units = find_tool_units(messages)
assert len(units) == 1
assistant_idx, response_indices = units[0]
assert assistant_idx == 1
assert response_indices == [2]
def test_anthropic_format_orphaned_tool_result(self):
"""Anthropic tool_result without matching tool_use is not included."""
messages = [
{"role": "user", "content": "Hi"},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "orphan_toolu",
"content": "orphaned result",
}
],
},
{"role": "assistant", "content": "Hello!"},
]
units = find_tool_units(messages)
assert units == []
def test_mixed_openai_and_anthropic_formats(self):
"""Both OpenAI and Anthropic formats can coexist (edge case)."""
messages = [
{"role": "user", "content": "Do things"},
# OpenAI format
{
"role": "assistant",
"tool_calls": [
{"id": "call_1", "function": {"name": "openai_tool", "arguments": "{}"}}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "openai result"},
# Anthropic format
{
"role": "assistant",
"content": [
{"type": "tool_use", "id": "toolu_2", "name": "anthropic_tool", "input": {}}
],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_2", "content": "anthropic result"}
],
},
]
units = find_tool_units(messages)
assert len(units) == 2
# First unit: OpenAI format (assistant at 1, tool response at 2)
assert units[0] == (1, [2])
# Second unit: Anthropic format (assistant at 3, user with tool_result at 4)
assert units[1] == (3, [4])
# --- TestGetMessageContentText ---
class TestGetMessageContentText:
"""Tests for get_message_content_text function."""
def test_string_content(self):
"""Extracts string content directly."""
msg = {"role": "user", "content": "Hello, world!"}
text = get_message_content_text(msg)
assert text == "Hello, world!"
def test_list_content(self):
"""Extracts text from list content (multimodal)."""
msg = {
"role": "user",
"content": [
{"type": "text", "text": "First part"},
{"type": "image", "source": {}},
{"type": "text", "text": "Second part"},
],
}
text = get_message_content_text(msg)
assert "First part" in text
assert "Second part" in text
def test_none_content(self):
"""Returns empty string for None content."""
msg = {"role": "assistant", "content": None}
text = get_message_content_text(msg)
assert text == ""
def test_mixed_content_list(self):
"""Handles list with both dict and string items."""
msg = {
"role": "user",
"content": [
{"type": "text", "text": "Dict text"},
"Plain string",
],
}
text = get_message_content_text(msg)
assert "Dict text" in text
assert "Plain string" in text
def test_missing_content_key(self):
"""Returns empty string when content key is missing."""
msg = {"role": "user"}
text = get_message_content_text(msg)
assert text == ""
def test_non_text_type_skipped(self):
"""Non-text types in list are skipped."""
msg = {
"role": "user",
"content": [
{"type": "image", "data": "..."},
{"type": "text", "text": "Only this"},
],
}
text = get_message_content_text(msg)
assert text == "Only this"
def test_empty_list_content(self):
"""Empty list content returns empty string."""
msg = {"role": "user", "content": []}
text = get_message_content_text(msg)
assert text == ""
# --- Additional fixtures for complex tests ---
@pytest.fixture
def sample_messages():
"""Basic conversation messages."""
return [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you!"},
]
@pytest.fixture
def sample_messages_with_tools():
"""Conversation with tool calls and responses."""
return [
{"role": "system", "content": "You are a helpful assistant with tools."},
{"role": "user", "content": "Search for user 12345"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_123",
"type": "function",
"function": {"name": "search_user", "arguments": '{"user_id": "12345"}'},
}
],
},
{
"role": "tool",
"tool_call_id": "call_123",
"content": '{"id": "12345", "name": "Alice", "email": "alice@example.com"}',
},
{"role": "assistant", "content": "I found user Alice with ID 12345."},
]
# --- Anthropic tool_result content blocks (chopratejas/headroom#813) ---
@pytest.fixture
def big_json_payload():
"""JSON blob large enough to trip the json_bloat detector (>500 tokens)."""
return "{" + ",".join(f'"key_{i}": "value padding text {i}"' for i in range(200)) + "}"
class TestAnthropicToolResultBlocks:
"""Anthropic Messages format nests tool output in tool_result content blocks."""
def test_tool_result_with_nested_text_list(self, mock_tokenizer, big_json_payload):
message = {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_01",
"content": [{"type": "text", "text": big_json_payload}],
}
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
tool_blocks = [b for b in blocks if b.kind == "tool_result"]
assert len(tool_blocks) == 1
assert tool_blocks[0].text == big_json_payload
assert tool_blocks[0].flags["tool_call_id"] == "toolu_01"
assert tool_blocks[0].flags["waste_signals"]["json_bloat"] > 0
def test_tool_result_with_string_content(self, mock_tokenizer, big_json_payload):
message = {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_02",
"content": big_json_payload,
}
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
tool_blocks = [b for b in blocks if b.kind == "tool_result"]
assert len(tool_blocks) == 1
assert tool_blocks[0].text == big_json_payload
assert tool_blocks[0].flags["waste_signals"]["json_bloat"] > 0
def test_mixed_text_and_tool_result(self, mock_tokenizer, big_json_payload):
message = {
"role": "user",
"content": [
{"type": "text", "text": "Here is the output:"},
{
"type": "tool_result",
"tool_use_id": "toolu_03",
"content": [{"type": "text", "text": big_json_payload}],
},
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
assert [b.kind for b in blocks] == ["user", "tool_result"]
assert blocks[0].text == "Here is the output:"
assert blocks[1].text == big_json_payload
def test_empty_tool_result_content_emits_no_block(self, mock_tokenizer):
message = {
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_04", "content": []},
{"type": "tool_result", "tool_use_id": "toolu_05"},
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
# Nothing extractable: keep the container block so every message
# still yields at least one block.
assert [b.kind for b in blocks] == ["user"]
def test_tool_result_only_message_skips_container_block(self, mock_tokenizer, big_json_payload):
message = {
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_08", "content": big_json_payload}
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
assert [b.kind for b in blocks] == ["tool_result"]
def test_tool_result_dict_content_serialized_as_json(self, mock_tokenizer):
rows = {"rows": [{"id": i, "padding": "x" * 30} for i in range(120)]}
message = {
"role": "user",
"content": [{"type": "tool_result", "tool_use_id": "toolu_09", "content": rows}],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
tool_blocks = [b for b in blocks if b.kind == "tool_result"]
assert len(tool_blocks) == 1
assert tool_blocks[0].text.startswith('{"rows":')
assert tool_blocks[0].flags["waste_signals"]["json_bloat"] > 0
def test_missing_tool_use_id_yields_none_flag(self, mock_tokenizer, big_json_payload):
message = {
"role": "user",
"content": [{"type": "tool_result", "content": big_json_payload}],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
assert blocks[0].kind == "tool_result"
assert blocks[0].flags["tool_call_id"] is None
class TestStrandsToolResultBlocks:
"""Strands/Bedrock converse format: toolResult content parts."""
def test_strands_text_content(self, mock_tokenizer, big_json_payload):
message = {
"role": "user",
"content": [
{
"toolResult": {
"toolUseId": "strands_01",
"content": [{"text": big_json_payload}],
"status": "success",
}
}
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
assert [b.kind for b in blocks] == ["tool_result"]
assert blocks[0].text == big_json_payload
assert blocks[0].flags["tool_call_id"] == "strands_01"
assert blocks[0].flags["waste_signals"]["json_bloat"] > 0
def test_strands_json_content(self, mock_tokenizer):
rows = {"rows": [{"id": i, "padding": "x" * 30} for i in range(120)]}
message = {
"role": "user",
"content": [{"toolResult": {"toolUseId": "strands_02", "content": [{"json": rows}]}}],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
assert [b.kind for b in blocks] == ["tool_result"]
assert blocks[0].flags["waste_signals"]["json_bloat"] > 0
def test_non_text_inner_blocks_skipped(self, mock_tokenizer):
message = {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_06",
"content": [
{"type": "image", "source": {"type": "base64", "data": "abc"}},
{"type": "text", "text": "small result"},
"raw string piece",
],
}
],
}
blocks = parse_message_to_blocks(message, 0, mock_tokenizer)
tool_blocks = [b for b in blocks if b.kind == "tool_result"]
assert len(tool_blocks) == 1
assert tool_blocks[0].text == "small result\nraw string piece"
def test_parse_messages_aggregates_tool_result_waste(self, mock_tokenizer, big_json_payload):
messages = [
{"role": "user", "content": "Run the tool"},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_07",
"content": [{"type": "text", "text": big_json_payload}],
}
],
},
]
_, breakdown, waste = parse_messages(messages, mock_tokenizer)
assert waste.json_bloat_tokens > 0
assert breakdown["tool_result"] > 0
def test_waste_parity_with_openai_tool_role(self, mock_tokenizer, big_json_payload):
anthropic_messages = [
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "t1",
"content": [{"type": "text", "text": big_json_payload}],
}
],
}
]
openai_messages = [{"role": "tool", "tool_call_id": "t1", "content": big_json_payload}]
_, _, anthropic_waste = parse_messages(anthropic_messages, mock_tokenizer)
_, _, openai_waste = parse_messages(openai_messages, mock_tokenizer)
assert anthropic_waste.total() > 0
assert anthropic_waste.total() == openai_waste.total()
# --- TestCallArgMatchReread ---
class TestCallArgMatchReread:
"""Tests for re-issued-call (arg-match) reread detection in parse_messages."""
LARGE_CONTENT = "def handler(event):\n return process(event)\n" * 10 # > 200 chars
CHANGED_CONTENT = LARGE_CONTENT + "# mtime 1718000000\n"
def _expected_tokens(self, text):
"""Mirror mock_tokenizer + message overhead used for tool_result blocks."""
return len(text) // 4 + 1 + 4
@staticmethod
def _filler(n):
"""Interleaved turns that push a repeat beyond the polling gap."""
return [
{"role": "assistant" if i % 2 == 0 else "user", "content": f"step {i} of the task"}
for i in range(n)
]
@staticmethod
def _openai_call(call_id, name, arguments):
return {
"role": "assistant",
"content": None,
"tool_calls": [{"id": call_id, "function": {"name": name, "arguments": arguments}}],
}
@staticmethod
def _openai_result(call_id, content):
return {"role": "tool", "tool_call_id": call_id, "content": content}
def test_canonical_call_key_normalizes_serialization(self):
"""Reordered JSON-string args, dict args, and spaced JSON hash equal."""
from headroom.parser import _canonical_call_key
k1 = _canonical_call_key("read_file", '{"path": "a.py", "lines": 100}')
k2 = _canonical_call_key("read_file", '{"lines":100,"path":"a.py"}')
k3 = _canonical_call_key("read_file", {"path": "a.py", "lines": 100})
assert k1 == k2 == k3
assert _canonical_call_key("read_file", '{"path": "b.py", "lines": 100}') != k1
assert _canonical_call_key("grep", '{"path": "a.py", "lines": 100}') != k1
def test_reissued_call_changed_result_counts(self, mock_tokenizer):
"""Identical call re-issued far apart counts even when result bytes differ."""
messages = (
[
self._openai_call("c1", "read_file", '{"path": "a.py", "lines": 100}'),
self._openai_result("c1", self.LARGE_CONTENT),
]
+ self._filler(4)
+ [
self._openai_call("c2", "read_file", '{"lines":100,"path":"a.py"}'),
self._openai_result("c2", self.CHANGED_CONTENT),
]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.CHANGED_CONTENT)
def test_identical_result_not_double_counted(self, mock_tokenizer):
"""Byte-identical repeat is counted once (content-hash pass wins)."""
messages = (
[
self._openai_call("c1", "read_file", '{"path": "a.py"}'),
self._openai_result("c1", self.LARGE_CONTENT),
]
+ self._filler(4)
+ [
self._openai_call("c2", "read_file", '{"path": "a.py"}'),
self._openai_result("c2", self.LARGE_CONTENT),
]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.LARGE_CONTENT)
def test_adjacent_reissue_is_polling(self, mock_tokenizer):
"""Back-to-back identical calls (poll loop) are not re-reads."""
messages = [
self._openai_call("c1", "check_ci", '{"run": 7}'),
self._openai_result("c1", self.LARGE_CONTENT),
self._openai_call("c2", "check_ci", '{"run": 7}'),
self._openai_result("c2", self.CHANGED_CONTENT),
]
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_different_args_not_matched(self, mock_tokenizer):
"""Same tool with different arguments is not a re-issued call."""
messages = (
[
self._openai_call("c1", "read_file", '{"path": "a.py"}'),
self._openai_result("c1", self.LARGE_CONTENT),
]
+ self._filler(4)
+ [
self._openai_call("c2", "read_file", '{"path": "b.py"}'),
self._openai_result("c2", self.CHANGED_CONTENT),
]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_small_result_ignored(self, mock_tokenizer):
"""Repeat of a call whose result is trivially small is skipped."""
messages = (
[
self._openai_call("c1", "run_tests", "{}"),
self._openai_result("c1", "ok"),
]
+ self._filler(4)
+ [
self._openai_call("c2", "run_tests", "{}"),
self._openai_result("c2", "ok again"),
]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_repeat_call_without_result_not_counted(self, mock_tokenizer):
"""A re-issued call with no recorded result contributes nothing."""
messages = (
[
self._openai_call("c1", "read_file", '{"path": "a.py"}'),
self._openai_result("c1", self.LARGE_CONTENT),
]
+ self._filler(4)
+ [self._openai_call("c2", "read_file", '{"path": "a.py"}')]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == 0
def test_anthropic_tool_use_produces_tool_call_blocks(self, mock_tokenizer):
"""Anthropic tool_use parts become tool_call blocks with call metadata."""
messages = [
{
"role": "assistant",
"content": [
{"type": "text", "text": "Reading the file now."},
{
"type": "tool_use",
"id": "t1",
"name": "read_file",
"input": {"path": "a.py"},
},
],
}
]
blocks, _, _ = parse_messages(messages, mock_tokenizer)
tool_calls = [b for b in blocks if b.kind == "tool_call"]
assert len(tool_calls) == 1
assert tool_calls[0].flags["function_name"] == "read_file"
assert tool_calls[0].flags["tool_call_id"] == "t1"
assert tool_calls[0].flags["call_key"]
def test_anthropic_reissued_call_changed_result_counts(self, mock_tokenizer):
"""Full Anthropic-format flow: re-issued tool_use with drifted result."""
def call(uid):
return {
"role": "assistant",
"content": [
{"type": "tool_use", "id": uid, "name": "read_file", "input": {"path": "a.py"}}
],
}
def result(uid, content):
return {
"role": "user",
"content": [{"type": "tool_result", "tool_use_id": uid, "content": content}],
}
messages = (
[call("t1"), result("t1", self.LARGE_CONTENT)]
+ self._filler(4)
+ [call("t2"), result("t2", self.CHANGED_CONTENT)]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.CHANGED_CONTENT)
def test_strands_tooluse_matched(self, mock_tokenizer):
"""Strands/Bedrock toolUse/toolResult format is matched the same way."""
def call(uid):
return {
"role": "assistant",
"content": [{"toolUse": {"toolUseId": uid, "name": "search", "input": {"q": "x"}}}],
}
def result(uid, content):
return {
"role": "user",
"content": [{"toolResult": {"toolUseId": uid, "content": [{"text": content}]}}],
}
messages = (
[call("s1"), result("s1", self.LARGE_CONTENT)]
+ self._filler(4)
+ [call("s2"), result("s2", self.CHANGED_CONTENT)]
)
_, _, waste = parse_messages(messages, mock_tokenizer)
assert waste.reread_tokens == self._expected_tokens(self.CHANGED_CONTENT)
def test_cross_format_call_key_parity(self, mock_tokenizer):
"""OpenAI JSON-string args and Anthropic dict input produce the same call_key."""
openai_msgs = [self._openai_call("c1", "read_file", '{"lines": 100, "path": "a.py"}')]
anthropic_msgs = [
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "t1",
"name": "read_file",
"input": {"path": "a.py", "lines": 100},
}
],
}
]
o_blocks, _, _ = parse_messages(openai_msgs, mock_tokenizer)
a_blocks, _, _ = parse_messages(anthropic_msgs, mock_tokenizer)
o_key = [b for b in o_blocks if b.kind == "tool_call"][0].flags["call_key"]
a_key = [b for b in a_blocks if b.kind == "tool_call"][0].flags["call_key"]
assert o_key == a_key