mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
feat(parser): detect re-issued identical tool calls as reread waste (#909)
Fixes #908 ## Problem Reread waste detection matches `tool_result` blocks by exact `content_hash` only. Two gaps hide a common waste pattern — the agent re-issuing the *same tool call* and paying full price for a near-identical result: 1. **Byte-different results escape matching.** Same tool, same arguments, but the second result differs trivially (embedded mtimes, timestamps, ordering) → different hash, zero reread counted. 2. **Anthropic `tool_use` parts were dropped entirely** in `parse_message_to_blocks` — only OpenAI-style `message.tool_calls` produced `tool_call` blocks, so Anthropic/Strands traffic had no call-side record at all. ## Fix - Parse Anthropic `tool_use` / Strands `toolUse` content parts into `tool_call` blocks (same shape as the OpenAI path: `function_name`, `tool_call_id` flags). - Tag every `tool_call` block with a canonical `call_key` = hash(name + arguments re-serialized with sorted keys), so `'{"path": "a.py", "lines": 100}'` (OpenAI JSON string) and `{"lines": 100, "path": "a.py"}` (Anthropic dict) hash equal — covered by a cross-format parity test. - Second reread pass in `parse_messages` groups calls by `call_key`: repeat invocations past the existing `REREAD_ADJACENT_GAP` polling guard count their **result** tokens into `reread_tokens`, subject to the existing `REREAD_MIN_TOKENS` floor. Results already counted by the content-hash pass are skipped, so byte-identical repeats are never double-counted. No new `WasteSignals` field — a byte-different re-fetch of an identical call is reread waste by the existing definition. Detection is Python-only (`parser.py`); no Rust parity surface. ## Proof Re-reading the same file twice, 7 messages apart, second serve differing only by an mtime line: ``` main: tool_call blocks: 2, reread_tokens: 0 this branch: tool_call blocks: 2, reread_tokens: 381 ``` ## Testing - 11 new tests (`TestCallArgMatchReread`): changed-result repeat counted (OpenAI + Anthropic + Strands formats), byte-identical repeat counted exactly once, polling gap skipped, different args not matched, sub-floor results skipped, repeat without result ignored, canonical-key normalization, cross-format call_key parity. - Full `tests/test_parser.py` suite: 87 passed. Consumer regression sweep (reporting, config, request outcome, read lifecycle, observability, storage): 188 passed. - `ruff check` + `ruff format --check` + `mypy headroom/parser.py` clean. Co-authored-by: integration-check <integration@local>
This commit is contained in:
parent
0632eba6c3
commit
7d4ae86ec0
2 changed files with 319 additions and 0 deletions
|
|
@ -48,6 +48,22 @@ def compute_hash(text: str) -> str:
|
|||
return hashlib.md5(text.encode()).hexdigest()[:16] # nosec B324
|
||||
|
||||
|
||||
def _canonical_call_key(name: str, arguments: Any) -> str:
|
||||
"""Canonical identity for a tool invocation: name + arguments with JSON
|
||||
key order normalized, so semantically identical calls hash equal even
|
||||
when the provider serializes arguments differently."""
|
||||
if isinstance(arguments, str):
|
||||
try:
|
||||
arguments = json.loads(arguments)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
if isinstance(arguments, (dict, list)):
|
||||
canon = json.dumps(arguments, sort_keys=True, separators=(",", ":"), default=str)
|
||||
else:
|
||||
canon = str(arguments)
|
||||
return compute_hash(f"{name}\x00{canon}")
|
||||
|
||||
|
||||
def _extract_tool_result_text(payload: dict[str, Any]) -> str:
|
||||
"""Extract text from a tool result payload.
|
||||
|
||||
|
|
@ -157,6 +173,7 @@ def parse_message_to_blocks(
|
|||
content = message.get("content")
|
||||
if content:
|
||||
tool_result_parts: list[dict[str, Any]] = []
|
||||
tool_use_parts: list[dict[str, Any]] = []
|
||||
if isinstance(content, str):
|
||||
text = content
|
||||
elif isinstance(content, list):
|
||||
|
|
@ -172,6 +189,13 @@ def parse_message_to_blocks(
|
|||
elif isinstance(part, dict) and "toolResult" in part:
|
||||
# Strands/Bedrock converse format; same treatment.
|
||||
tool_result_parts.append(part)
|
||||
elif isinstance(part, dict) and part.get("type") == "tool_use":
|
||||
# Anthropic Messages format: call side of the tool unit;
|
||||
# collect for dedicated tool_call blocks below.
|
||||
tool_use_parts.append(part)
|
||||
elif isinstance(part, dict) and "toolUse" in part:
|
||||
# Strands/Bedrock converse format; same treatment.
|
||||
tool_use_parts.append(part)
|
||||
elif isinstance(part, str):
|
||||
text_parts.append(part)
|
||||
text = "\n".join(text_parts)
|
||||
|
|
@ -242,6 +266,33 @@ def parse_message_to_blocks(
|
|||
)
|
||||
blocks.extend(tr_blocks)
|
||||
|
||||
for part in tool_use_parts:
|
||||
payload = part["toolUse"] if "toolUse" in part else part
|
||||
if not isinstance(payload, dict):
|
||||
continue
|
||||
tu_name = payload.get("name") or "unknown"
|
||||
tu_args = payload.get("input", {})
|
||||
tu_id = payload.get("toolUseId") if "toolUse" in part else payload.get("id")
|
||||
try:
|
||||
tu_args_text = json.dumps(tu_args, sort_keys=True, default=str)
|
||||
except (TypeError, ValueError):
|
||||
tu_args_text = str(tu_args)
|
||||
tu_text = f"{tu_name}({tu_args_text})"
|
||||
blocks.append(
|
||||
Block(
|
||||
kind="tool_call",
|
||||
text=tu_text,
|
||||
tokens_est=tokenizer.count_text(tu_text) + 10,
|
||||
content_hash=compute_hash(tu_text),
|
||||
source_index=index,
|
||||
flags={
|
||||
"tool_call_id": tu_id,
|
||||
"function_name": tu_name,
|
||||
"call_key": _canonical_call_key(tu_name, tu_args),
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
# Handle tool calls (assistant messages with tool_calls)
|
||||
tool_calls = message.get("tool_calls")
|
||||
if tool_calls:
|
||||
|
|
@ -259,6 +310,9 @@ def parse_message_to_blocks(
|
|||
flags={
|
||||
"tool_call_id": tc.get("id"),
|
||||
"function_name": func.get("name"),
|
||||
"call_key": _canonical_call_key(
|
||||
func.get("name") or "unknown", func.get("arguments", "")
|
||||
),
|
||||
},
|
||||
)
|
||||
)
|
||||
|
|
@ -315,6 +369,7 @@ def parse_messages(
|
|||
# at more than one position means the agent re-fetched something already
|
||||
# in context — an over-compression signal (#853). The first serve is
|
||||
# free; every repeat is counted as waste.
|
||||
counted_results: set[int] = set()
|
||||
reread_groups: dict[str, list[Block]] = {}
|
||||
for block in all_blocks:
|
||||
if block.kind == "tool_result" and block.tokens_est >= REREAD_MIN_TOKENS:
|
||||
|
|
@ -337,6 +392,44 @@ def parse_messages(
|
|||
prev_index = block.source_index
|
||||
if not is_polling:
|
||||
total_waste.reread_tokens += block.tokens_est
|
||||
counted_results.add(id(block))
|
||||
|
||||
# Re-issued-call detection: the agent invoking the same tool with the
|
||||
# same arguments again is a re-fetch even when the result bytes differ
|
||||
# (timestamps, mtimes, ordering defeat the content-hash pass above).
|
||||
# Same polling guard and size floor as above, applied to the repeat
|
||||
# invocation's result; results the content-hash pass already counted
|
||||
# are skipped so identical-content repeats are never counted twice.
|
||||
results_by_call_id: dict[str, Block] = {}
|
||||
for block in all_blocks:
|
||||
if block.kind == "tool_result":
|
||||
tc_id = block.flags.get("tool_call_id")
|
||||
if tc_id and tc_id not in results_by_call_id:
|
||||
results_by_call_id[tc_id] = block
|
||||
|
||||
call_groups: dict[str, list[Block]] = {}
|
||||
for block in all_blocks:
|
||||
if block.kind == "tool_call":
|
||||
call_key = block.flags.get("call_key")
|
||||
if call_key:
|
||||
call_groups.setdefault(call_key, []).append(block)
|
||||
|
||||
for group in call_groups.values():
|
||||
prev_index = group[0].source_index
|
||||
for block in group:
|
||||
if block.source_index == prev_index:
|
||||
continue
|
||||
is_polling = block.source_index - prev_index <= REREAD_ADJACENT_GAP
|
||||
prev_index = block.source_index
|
||||
if is_polling:
|
||||
continue
|
||||
result = results_by_call_id.get(block.flags.get("tool_call_id") or "")
|
||||
if result is None or result.tokens_est < REREAD_MIN_TOKENS:
|
||||
continue
|
||||
if id(result) in counted_results:
|
||||
continue
|
||||
total_waste.reread_tokens += result.tokens_est
|
||||
counted_results.add(id(result))
|
||||
|
||||
# Compute block breakdown
|
||||
breakdown: dict[str, int] = {}
|
||||
|
|
|
|||
|
|
@ -1051,3 +1051,229 @@ class TestStrandsToolResultBlocks:
|
|||
|
||||
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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue