diff --git a/headroom/config.py b/headroom/config.py index 32761d836..2b01d6bd6 100644 --- a/headroom/config.py +++ b/headroom/config.py @@ -3,6 +3,7 @@ from __future__ import annotations import fnmatch +import json from collections.abc import Iterable from dataclasses import InitVar, dataclass, field from datetime import datetime @@ -282,6 +283,37 @@ def _tool_name_aliases(name: str) -> tuple[str, ...]: return tuple(dict.fromkeys(aliases)) +# Hermes Agent's deferred-tool bridge. Hermes loads on-demand tools via a +# `tool_search`/`tool_describe`/`tool_call` indirection; on the wire the +# emitted tool call is named `tool_call` and the REAL tool name lives in the +# arguments payload (`{"name": "...", "arguments": {...}}`). Tool exclusion / +# protect lists match on the real name, so we must unwrap this bridge before +# building the tool_call_id -> name map, or whitelists silently no-op for all +# deferred tools. +_HERMES_TOOL_CALL_WRAPPER = "tool_call" + + +def unwrap_tool_call_name(name: str, arguments: Any) -> str: + """Extract the real tool name from a Hermes deferred ``tool_call`` wrapper. + + Non-wrapper names pass through unchanged. Malformed/unparseable wrappers + fail open and return the wrapper name (caller decides what that means). + """ + if name != _HERMES_TOOL_CALL_WRAPPER: + return name + raw = arguments + if isinstance(raw, str): + try: + raw = json.loads(raw) + except (ValueError, TypeError): + return name + if isinstance(raw, dict): + inner = raw.get("name") + if isinstance(inner, str) and inner.strip(): + return inner.strip() + return name + + def is_tool_excluded(name: str, exclude_tools: Iterable[str]) -> bool: """Return True if ``name`` matches the tool-exclusion set. diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index ffe770ef5..dd12a2290 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -44,6 +44,7 @@ if TYPE_CHECKING: import httpx from headroom.agent_savings import proxy_pipeline_kwargs +from headroom.config import unwrap_tool_call_name from headroom.copilot_auth import ( apply_copilot_api_auth, build_copilot_upstream_url, @@ -1638,6 +1639,10 @@ class OpenAIHandlerMixin: continue name = item.get("name") call_id = item.get("call_id") + if name: + # Hermes deferred tools arrive wrapped as `tool_call` with + # the real name inside the arguments/input payload. + name = unwrap_tool_call_name(name, item.get("arguments") or item.get("input")) if isinstance(name, str) and isinstance(call_id, str) and call_id: function_name_by_call_id[call_id] = name if isinstance(name, str) and ( diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 242e0aaa2..4aeb03f3e 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -57,6 +57,7 @@ from ..config import ( RelevanceScorerConfig, TransformResult, is_tool_excluded, + unwrap_tool_call_name, ) from ..parser import CCR_RETRIEVAL_MARKER_RE from ..tokenizer import Tokenizer @@ -4399,6 +4400,10 @@ class ContentRouter(Transform): tc_id = tc.get("id", "") fn = tc.get("function", {}) name = fn.get("name", "") + if name: + # Hermes deferred tools arrive wrapped as `tool_call` + # with the real name inside the arguments payload. + name = unwrap_tool_call_name(name, fn.get("arguments")) if tc_id and name: mapping[tc_id] = name args = _tool_call_args_text(fn.get("arguments")) @@ -4415,6 +4420,10 @@ class ContentRouter(Transform): if isinstance(block, dict) and block.get("type") == "tool_use": tc_id = block.get("id", "") name = block.get("name", "") + if name: + # Hermes deferred tools arrive wrapped as `tool_call` + # with the real name inside the input payload. + name = unwrap_tool_call_name(name, block.get("input")) if tc_id and name: mapping[tc_id] = name args = _tool_call_args_text(block.get("input")) diff --git a/tests/test_hermes_tool_call_unwrap.py b/tests/test_hermes_tool_call_unwrap.py new file mode 100644 index 000000000..b5c393989 --- /dev/null +++ b/tests/test_hermes_tool_call_unwrap.py @@ -0,0 +1,191 @@ +"""Tests for Hermes deferred-tool (`tool_call` wrapper) unwrapping. + +Hermes Agent loads on-demand tools via a `tool_search`/`tool_describe`/ +`tool_call` indirection: on the wire the emitted tool call is named +`tool_call` and the REAL tool name lives in the arguments payload +(`{"name": "...", "arguments": {...}}`). Tool exclusion / protect lists +match on the real name, so `_build_tool_name_map` must unwrap the bridge +or whitelists silently no-op for all deferred tools. + +These tests pin the `unwrap_tool_call_name` helper and its integration +into `ContentRouter._build_tool_name_map` (OpenAI + Anthropic paths). +""" + +from __future__ import annotations + +from headroom.config import ( + DEFAULT_EXCLUDE_TOOLS, + is_tool_excluded, + unwrap_tool_call_name, +) +from headroom.transforms.content_router import ContentRouter, ContentRouterConfig + +# --------------------------------------------------------------------------- +# Helper unit tests +# --------------------------------------------------------------------------- + + +def test_unwrap_passthrough_plain_name() -> None: + assert unwrap_tool_call_name("read_file", '{"path": "/x"}') == "read_file" + + +def test_unwrap_passthrough_none_arguments() -> None: + assert unwrap_tool_call_name("tool_call", None) == "tool_call" + + +def test_unwrap_passthrough_bad_json() -> None: + assert unwrap_tool_call_name("tool_call", "bad json") == "tool_call" + + +def test_unwrap_passthrough_missing_name_key() -> None: + assert unwrap_tool_call_name("tool_call", '{"no_name": true}') == "tool_call" + + +def test_unwrap_passthrough_empty_name() -> None: + assert unwrap_tool_call_name("", None) == "" + + +def test_unwrap_web_search() -> None: + assert ( + unwrap_tool_call_name("tool_call", '{"name": "web_search", "arguments": {}}') + == "web_search" + ) + + +def test_unwrap_read_file() -> None: + assert ( + unwrap_tool_call_name("tool_call", '{"name": "read_file", "arguments": {"path": "/x"}}') + == "read_file" + ) + + +def test_unwrap_mcp_tool() -> None: + assert ( + unwrap_tool_call_name( + "tool_call", '{"name": "mcp__codebase_memory__search", "arguments": {}}' + ) + == "mcp__codebase_memory__search" + ) + + +def test_unwrap_dict_arguments_form() -> None: + """Arguments may arrive as a dict (not JSON string) on some paths.""" + assert ( + unwrap_tool_call_name("tool_call", {"name": "search_files", "arguments": {"pattern": "x"}}) + == "search_files" + ) + + +def test_unwrap_whitelist_activation() -> None: + """Unwrapped names must activate the DEFAULT_EXCLUDE_TOOLS whitelist.""" + assert is_tool_excluded("web_search", DEFAULT_EXCLUDE_TOOLS) is True + unwrapped = unwrap_tool_call_name("tool_call", '{"name": "web_search", "arguments": {}}') + assert is_tool_excluded(unwrapped, DEFAULT_EXCLUDE_TOOLS) is True + + +# --------------------------------------------------------------------------- +# _build_tool_name_map integration tests +# --------------------------------------------------------------------------- + + +def _router(exclude_tools: set[str] | None = None) -> ContentRouter: + config = ContentRouterConfig( + min_section_tokens=10, + enable_kompress=False, + exclude_tools=exclude_tools, + ) + return ContentRouter(config) + + +def test_build_tool_name_map_openai_wrapped() -> None: + """OpenAI-format assistant tool_calls with Hermes tool_call wrapper.""" + messages = [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_wrapped_1", + "type": "function", + "function": { + "name": "tool_call", + "arguments": '{"name": "read_file", "arguments": {"path": "/x"}}', + }, + }, + { + "id": "call_plain_2", + "type": "function", + "function": {"name": "web_search", "arguments": '{"query": "q"}'}, + }, + ], + } + ] + router = _router() + mapping = router._build_tool_name_map(messages) + assert mapping["call_wrapped_1"] == "read_file", ( + "wrapped tool_call must map to the real tool name" + ) + assert mapping["call_plain_2"] == "web_search", "plain tool names must pass through unchanged" + + +def test_build_tool_name_map_anthropic_wrapped() -> None: + """Anthropic-format tool_use blocks with Hermes tool_call wrapper.""" + messages = [ + { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "toolu_wrapped_1", + "name": "tool_call", + "input": {"name": "headroom_retrieve", "arguments": {"hash": "abc"}}, + }, + { + "type": "tool_use", + "id": "toolu_plain_2", + "name": "Read", + "input": {"file_path": "/x"}, + }, + ], + } + ] + router = _router() + mapping = router._build_tool_name_map(messages) + assert mapping["toolu_wrapped_1"] == "headroom_retrieve", ( + "wrapped tool_call must map to the real tool name" + ) + assert mapping["toolu_plain_2"] == "Read", "plain tool names must pass through unchanged" + + +def test_build_tool_name_map_wrapped_not_excluded_before_unwrap() -> None: + """Sanity: without unwrapping, a wrapped read_file is NOT excluded. + + This documents the failure mode the fix addresses: `tool_call` is not in + DEFAULT_EXCLUDE_TOOLS, so a whitelist match would never fire. + """ + assert is_tool_excluded("tool_call", DEFAULT_EXCLUDE_TOOLS) is False + assert is_tool_excluded("read_file", DEFAULT_EXCLUDE_TOOLS) is False + + +def test_build_tool_name_map_exclusion_after_unwrap() -> None: + """Unwrapped names feed is_tool_excluded for whitelist decisions.""" + router = _router(exclude_tools=set(DEFAULT_EXCLUDE_TOOLS) | {"read_file"}) + messages = [ + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_rf_1", + "type": "function", + "function": { + "name": "tool_call", + "arguments": '{"name": "read_file", "arguments": {"path": "/x"}}', + }, + } + ], + } + ] + mapping = router._build_tool_name_map(messages) + assert mapping["call_rf_1"] == "read_file" + assert is_tool_excluded(mapping["call_rf_1"], router.config.exclude_tools or set()) is True