refactor(display): one think-tag list; ACP tool titles derive from agent/display previews
The CLI stream mixin and the gateway think filter each carried a hand-copied think-tag tuple guarded by a "must stay in sync" comment; adding a tag meant three edits. The scrubber (agent/think_scrubber.py) now exports THINK_OPEN_TAGS/THINK_CLOSE_TAGS and both consumers (and strip_think_blocks' regexes) bind to them. acp_adapter/tools.py::_TITLE_BUILDERS hand-rolled 25 per-tool titles that agent/display.build_tool_preview already produces (with redaction). ACP titles are now "<tool>: <preview>"; no ACP-specific overrides remained necessary.
This commit is contained in:
teknium1
@@ -10,6 +10,8 @@ from typing import Any, Callable, Dict, List, Optional
|
||||
import acp
|
||||
from acp.schema import ToolCallLocation, ToolCallProgress, ToolCallStart, ToolKind
|
||||
|
||||
from agent.display import build_tool_preview
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Hermes tool name -> ACP ToolKind (anything unlisted is "other").
|
||||
@@ -84,11 +86,6 @@ def _first(data: Args, *keys: str, default: Any = "") -> Any:
|
||||
return next((data[k] for k in keys if data.get(k)), default)
|
||||
|
||||
|
||||
def _clip(text: str, limit: int) -> str:
|
||||
"""Hard-truncate to ``limit`` chars with a trailing ellipsis."""
|
||||
return text if len(text) <= limit else text[: limit - 3] + "..."
|
||||
|
||||
|
||||
def _fmt(value: Any, template: str, fallback: str) -> str:
|
||||
"""``template.format(value)`` when value is truthy, else ``fallback``."""
|
||||
return template.format(value) if value else fallback
|
||||
@@ -193,68 +190,12 @@ def _tool_result_failed(result: Optional[str], tool_name: str | None = None) ->
|
||||
# --- tool-call titles -------------------------------------------------------
|
||||
|
||||
|
||||
def _title_web_extract(args: Args) -> str:
|
||||
urls = args.get("urls", [])
|
||||
if not urls:
|
||||
return "web extract"
|
||||
first = urls[0]
|
||||
if isinstance(first, dict):
|
||||
first = first.get("url") or first.get("href") or "?"
|
||||
elif not isinstance(first, str):
|
||||
first = "?"
|
||||
return f"extract: {first}" + (f" (+{len(urls)-1})" if len(urls) > 1 else "")
|
||||
|
||||
|
||||
def _title_delegate(args: Args) -> str:
|
||||
if isinstance(tasks := args.get("tasks"), list) and tasks:
|
||||
return f"delegate batch ({len(tasks)} tasks)"
|
||||
return f"delegate: {_clip(goal, 60)}" if (goal := args.get("goal", "")) else "delegate task"
|
||||
|
||||
|
||||
def _title_execute_code(args: Args) -> str:
|
||||
first_line = next((line.strip() for line in _arg(args, "code").splitlines() if line.strip()), "")
|
||||
return _fmt(_clip(first_line, 70), "python: {}", "python code")
|
||||
|
||||
|
||||
def _title_skill_manage(args: Args) -> str:
|
||||
name, file_path = _arg(args, "name", default="?"), _arg(args, "file_path")
|
||||
target = _clip(f"{name}/{file_path}" if file_path else name, 64)
|
||||
return f"skill {_arg(args, 'action', default='manage')}: {target}"
|
||||
|
||||
|
||||
_TITLE_BUILDERS: Dict[str, Callable[[Args], str]] = {
|
||||
"terminal": lambda a: f"terminal: {_clip(a.get('command', ''), 80)}",
|
||||
"read_file": lambda a: f"read: {a.get('path', '?')}",
|
||||
"write_file": lambda a: f"write: {a.get('path', '?')}",
|
||||
"patch": lambda a: f"patch ({a.get('mode', 'replace')}): {a.get('path', '?')}",
|
||||
"search_files": lambda a: f"search: {a.get('pattern', '?')}",
|
||||
"web_search": lambda a: f"web search: {a.get('query', '?')}",
|
||||
"web_extract": _title_web_extract,
|
||||
"process": lambda a: _fmt(_arg(a, "session_id"), f"process {_arg(a, 'action', default='manage')}: {{}}",
|
||||
f"process {_arg(a, 'action', default='manage')}"),
|
||||
"delegate_task": _title_delegate,
|
||||
"session_search": lambda a: _fmt(_arg(a, "query"), "session search: {}", "recent sessions"),
|
||||
"memory": lambda a: f"memory {_arg(a, 'action', default='manage')}: {_arg(a, 'target', default='memory')}",
|
||||
"execute_code": _title_execute_code,
|
||||
"todo": lambda a: f"todo ({_plural(len(a['todos']), 'item')})" if isinstance(a.get("todos"), list) else "todo",
|
||||
"skill_view": lambda a: f"skill view ({_arg(a, 'name', default='?')}{_fmt(_arg(a, 'file_path'), '/{}', '')})",
|
||||
"skills_list": lambda a: _fmt(_arg(a, "category"), "skills list ({})", "skills list"),
|
||||
"skill_manage": _title_skill_manage,
|
||||
"browser_navigate": lambda a: f"navigate: {a.get('url', '?')}",
|
||||
"browser_snapshot": lambda a: "browser snapshot",
|
||||
"browser_vision": lambda a: f"browser vision: {str(a.get('question', '?'))[:50]}",
|
||||
"browser_get_images": lambda a: "browser images",
|
||||
"vision_analyze": lambda a: f"analyze image: {str(a.get('question', '?'))[:50]}",
|
||||
"image_generate": lambda a: _fmt(_arg(a, "prompt", "description")[:50], "generate image: {}", "generate image"),
|
||||
"cronjob": lambda a: _fmt(_arg(a, "job_id", "id"), f"cron {_arg(a, 'action', default='manage')}: {{}}",
|
||||
f"cron {_arg(a, 'action', default='manage')}"),
|
||||
}
|
||||
|
||||
|
||||
def build_tool_title(tool_name: str, args: Args) -> str:
|
||||
"""Build a human-readable title for a tool call (defaults to the tool name)."""
|
||||
builder = _TITLE_BUILDERS.get(tool_name)
|
||||
return builder(args) if builder is not None else tool_name
|
||||
"""``<tool_name>: <preview>`` using the same per-tool preview (and argument redaction) as
|
||||
every other Hermes surface, so ACP clients never show a different summary than the CLI/TUI;
|
||||
bare tool name when the arguments yield no preview."""
|
||||
preview = build_tool_preview(tool_name, args, max_len=80)
|
||||
return f"{tool_name}: {preview}" if preview else tool_name
|
||||
|
||||
|
||||
# --- completion formatters; all share the signature (tool_name, result, args) --
|
||||
|
||||
@@ -21,6 +21,7 @@ from agent.message_sanitization import (
|
||||
)
|
||||
from agent.prompt_builder import STEER_DISPLAY_KIND, steer_user_row
|
||||
from agent.tool_dispatch_helpers import _trajectory_normalize_msg, make_tool_result_message
|
||||
from agent.think_scrubber import THINK_TAG_NAMES
|
||||
from agent.trajectory import convert_scratchpad_to_think
|
||||
from agent.credential_pool import (
|
||||
STATUS_EXHAUSTED, credential_pool_matches_provider, resolve_runtime_pool_key
|
||||
@@ -33,10 +34,9 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
# Cap same-entry OAuth refreshes on a persistent auth failure, else a single-entry pool re-mints forever.
|
||||
_MAX_AUTH_REFRESH_ATTEMPTS = 2
|
||||
_REASONING_TAG_NAMES = ("think", "thinking", "reasoning", "REASONING_SCRATCHPAD", "thought")
|
||||
_TOOL_CALL_TAG_NAMES = ("tool_call", "tool_calls", "tool_result", "function_call", "function_calls")
|
||||
_REASONING_BLOCK_PATTERNS = tuple(
|
||||
re.compile(rf"<{name}>.*?</{name}>", re.DOTALL | re.IGNORECASE) for name in _REASONING_TAG_NAMES
|
||||
re.compile(rf"<{name}>.*?</{name}>", re.DOTALL | re.IGNORECASE) for name in THINK_TAG_NAMES
|
||||
)
|
||||
_TOOL_CALL_BLOCK_PATTERNS = tuple(
|
||||
re.compile(rf"<{name}\b[^>]*>.*?</{name}>", re.DOTALL | re.IGNORECASE)
|
||||
@@ -50,10 +50,10 @@ _NAMED_FUNCTION_BLOCK_PATTERN = re.compile(
|
||||
r'(?:(?:(?!</function>).)*)</function>', re.DOTALL | re.IGNORECASE,
|
||||
)
|
||||
_UNTERMINATED_REASONING_BLOCK_PATTERN = re.compile(
|
||||
rf'(?:^|\n)[ \t]*<(?:{"|".join(_REASONING_TAG_NAMES)})\b[^>]*>.*$', re.DOTALL | re.IGNORECASE
|
||||
rf'(?:^|\n)[ \t]*<(?:{"|".join(THINK_TAG_NAMES)})\b[^>]*>.*$', re.DOTALL | re.IGNORECASE
|
||||
)
|
||||
_ORPHAN_REASONING_TAG_PATTERN = re.compile(
|
||||
rf'</?(?:{"|".join(_REASONING_TAG_NAMES)})>\s*', re.IGNORECASE
|
||||
rf'</?(?:{"|".join(THINK_TAG_NAMES)})>\s*', re.IGNORECASE
|
||||
)
|
||||
_STRAY_TOOL_CALL_CLOSER_PATTERN = re.compile(
|
||||
rf'</(?:{"|".join(_TOOL_CALL_TAG_NAMES)}|function)>\s*', re.IGNORECASE
|
||||
@@ -1207,7 +1207,7 @@ _TRANSIENT_TRANSPORT_ERRORS = frozenset({
|
||||
})
|
||||
_INLINE_REASONING_PATTERNS = tuple(
|
||||
re.compile(rf"<{tag}>(.*?)</{tag}>", re.DOTALL | re.IGNORECASE)
|
||||
for tag in ("think", "thinking", "thought", "reasoning", "REASONING_SCRATCHPAD")
|
||||
for tag in THINK_TAG_NAMES
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -13,7 +13,15 @@ from __future__ import annotations
|
||||
import re
|
||||
from typing import Tuple
|
||||
|
||||
__all__ = ["StreamingThinkScrubber"]
|
||||
__all__ = ["StreamingThinkScrubber", "THINK_TAG_NAMES", "THINK_OPEN_TAGS", "THINK_CLOSE_TAGS"]
|
||||
|
||||
# The one list of model reasoning tag names. Every surface that hides reasoning (this scrubber,
|
||||
# the CLI stream filter, the gateway stream filter, the final-response regex stripper) binds to
|
||||
# these; a tag added here is covered everywhere. Consumers match case-insensitively, so the
|
||||
# literal tags are lowercase.
|
||||
THINK_TAG_NAMES: Tuple[str, ...] = ("think", "thinking", "reasoning", "thought", "REASONING_SCRATCHPAD")
|
||||
THINK_OPEN_TAGS: Tuple[str, ...] = tuple(f"<{name.lower()}>" for name in THINK_TAG_NAMES)
|
||||
THINK_CLOSE_TAGS: Tuple[str, ...] = tuple(f"</{name.lower()}>" for name in THINK_TAG_NAMES)
|
||||
|
||||
|
||||
class StreamingThinkScrubber:
|
||||
@@ -24,11 +32,9 @@ class StreamingThinkScrubber:
|
||||
was emitted yet — decides whether an open tag at buffer position 0 sits at a block boundary).
|
||||
"""
|
||||
|
||||
_OPEN_TAG_NAMES: Tuple[str, ...] = ("think", "thinking", "reasoning", "thought", "REASONING_SCRATCHPAD")
|
||||
|
||||
# Lowercased literal tags so the hot path does string ops, not regex per feed().
|
||||
_OPEN_TAGS: Tuple[str, ...] = tuple(f"<{name.lower()}>" for name in _OPEN_TAG_NAMES)
|
||||
_CLOSE_TAGS: Tuple[str, ...] = tuple(f"</{name.lower()}>" for name in _OPEN_TAG_NAMES)
|
||||
# Literal tags so the hot path does string ops, not regex per feed().
|
||||
_OPEN_TAGS: Tuple[str, ...] = THINK_OPEN_TAGS
|
||||
_CLOSE_TAGS: Tuple[str, ...] = THINK_CLOSE_TAGS
|
||||
_ALL_TAGS: Tuple[str, ...] = _OPEN_TAGS + _CLOSE_TAGS
|
||||
_MAX_TAG_LEN: int = max(len(tag) for tag in _ALL_TAGS)
|
||||
# Orphan close tag plus trailing whitespace (matches _strip_think_blocks case 3).
|
||||
|
||||
@@ -9,6 +9,7 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from agent.think_scrubber import THINK_CLOSE_TAGS, THINK_OPEN_TAGS
|
||||
from agent.think_scrubber import StreamingThinkScrubber as _Scrubber
|
||||
|
||||
logger = logging.getLogger("gateway.stream_consumer")
|
||||
@@ -17,21 +18,13 @@ logger = logging.getLogger("gateway.stream_consumer")
|
||||
class StreamThinkFilterMixin:
|
||||
"""Progressive <think>-tag suppression over streamed deltas."""
|
||||
|
||||
# Must stay in sync with cli.py _OPEN_TAGS/_CLOSE_TAGS and
|
||||
# run_agent.py _strip_think_blocks() tag variants.
|
||||
_OPEN_THINK_TAGS = (
|
||||
"<REASONING_SCRATCHPAD>", "<think>", "<reasoning>",
|
||||
"<THINKING>", "<thinking>", "<thought>",
|
||||
)
|
||||
_CLOSE_THINK_TAGS = (
|
||||
"</REASONING_SCRATCHPAD>", "</think>", "</reasoning>",
|
||||
"</THINKING>", "</thinking>", "</thought>",
|
||||
)
|
||||
_OPEN_THINK_TAGS = THINK_OPEN_TAGS
|
||||
_CLOSE_THINK_TAGS = THINK_CLOSE_TAGS
|
||||
|
||||
def _at_block_boundary(self, buf: str, idx: int) -> bool:
|
||||
"""Tag at ``idx`` starts a block: start of text, or newline + optional whitespace.
|
||||
|
||||
Prose that merely *mentions* a tag must not trigger (mirrors cli.py).
|
||||
Prose that merely *mentions* a tag must not trigger.
|
||||
"""
|
||||
acc_boundary = not self._accumulated or self._accumulated.endswith("\n")
|
||||
if idx == 0:
|
||||
|
||||
@@ -17,11 +17,12 @@ from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from rich.markup import escape as _escape
|
||||
|
||||
from agent.think_scrubber import THINK_CLOSE_TAGS, THINK_OPEN_TAGS
|
||||
|
||||
# Model-generated reasoning tags: suppressed during streaming (they'd display as raw XML;
|
||||
# the agent strips them from final_response too) unless show_reasoning routes them to the box.
|
||||
_OPEN_TAGS = (
|
||||
"<REASONING_SCRATCHPAD>", "<think>", "<reasoning>", "<THINKING>", "<thinking>", "<thought>")
|
||||
_CLOSE_TAGS = tuple("</" + t[1:] for t in _OPEN_TAGS)
|
||||
_OPEN_TAGS = THINK_OPEN_TAGS
|
||||
_CLOSE_TAGS = THINK_CLOSE_TAGS
|
||||
_MAX_CLOSE_TAG_LEN = max(len(t) for t in _CLOSE_TAGS)
|
||||
|
||||
# Ordered (prefix, status) rows for _slow_command_status — first match wins.
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
"""Tests for acp_adapter.tools — tool kind mapping and ACP content building."""
|
||||
|
||||
|
||||
import pytest
|
||||
|
||||
from acp_adapter.edit_approval import EditProposal
|
||||
from acp_adapter.tools import (
|
||||
TOOL_KIND_MAP,
|
||||
@@ -88,30 +90,41 @@ class TestBuildToolTitle:
|
||||
|
||||
def test_read_file_title(self):
|
||||
title = build_tool_title("read_file", {"path": "/etc/hosts"})
|
||||
assert "/etc/hosts" in title
|
||||
|
||||
assert "hosts" in title
|
||||
|
||||
def test_search_title(self):
|
||||
title = build_tool_title("search_files", {"pattern": "TODO"})
|
||||
assert "TODO" in title
|
||||
|
||||
|
||||
|
||||
|
||||
def test_skill_view_title_includes_skill_name(self):
|
||||
title = build_tool_title("skill_view", {"name": "github-pitfalls"})
|
||||
assert title == "skill view (github-pitfalls)"
|
||||
|
||||
assert "github-pitfalls" in title
|
||||
|
||||
def test_execute_code_title_includes_first_code_line(self):
|
||||
title = build_tool_title("execute_code", {"code": "\nfrom hermes_tools import terminal\nprint('done')"})
|
||||
assert title == "python: from hermes_tools import terminal"
|
||||
|
||||
assert "from hermes_tools import terminal" in title
|
||||
|
||||
def test_unknown_tool_uses_name(self):
|
||||
title = build_tool_title("some_new_tool", {"foo": "bar"})
|
||||
assert title == "some_new_tool"
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tool_name, args",
|
||||
[
|
||||
("terminal", {"command": "git status --short"}),
|
||||
("read_file", {"path": "/etc/hosts", "offset": 10}),
|
||||
("search_files", {"pattern": "TODO", "path": "src"}),
|
||||
("web_search", {"query": "hermes agent acp"}),
|
||||
("execute_code", {"code": "\nfrom hermes_tools import terminal\nprint('done')"}),
|
||||
("skill_view", {"name": "github", "file_path": "references/x.md"}),
|
||||
],
|
||||
)
|
||||
def test_title_derives_from_display_preview(self, tool_name, args):
|
||||
"""ACP titles are the shared agent.display preview, not a parallel per-tool table."""
|
||||
from agent.display import build_tool_preview
|
||||
|
||||
assert build_tool_preview(tool_name, args, max_len=80) in build_tool_title(tool_name, args)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# build_tool_start
|
||||
@@ -166,7 +179,7 @@ class TestBuildToolStart:
|
||||
args = {"url": "https://x.com"}
|
||||
result = build_tool_start("tc-browser-start", "browser_navigate", args)
|
||||
assert isinstance(result, ToolCallStart)
|
||||
assert result.title == "navigate: https://x.com"
|
||||
assert "https://x.com" in result.title
|
||||
assert result.kind == "fetch"
|
||||
assert result.content[0].content.text == '{\n "url": "https://x.com"\n}'
|
||||
assert result.raw_input is None
|
||||
|
||||
@@ -6,7 +6,10 @@ from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from agent.think_scrubber import THINK_CLOSE_TAGS, THINK_OPEN_TAGS, THINK_TAG_NAMES
|
||||
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
|
||||
from gateway.stream_consumer_think import StreamThinkFilterMixin
|
||||
from hermes_cli import cli_stream_mixin
|
||||
|
||||
|
||||
def test_stream_send_metadata_carries_original_reply_anchor():
|
||||
@@ -931,6 +934,22 @@ class TestFilterAndAccumulate:
|
||||
assert c._accumulated == "Visible answer"
|
||||
assert "hidden reasoning" not in c._accumulated
|
||||
|
||||
def test_tag_lists_are_the_scrubbers(self):
|
||||
"""One owner: CLI and gateway stream filters bind the scrubber's tag tuples, not copies."""
|
||||
assert StreamThinkFilterMixin._OPEN_THINK_TAGS is THINK_OPEN_TAGS
|
||||
assert StreamThinkFilterMixin._CLOSE_THINK_TAGS is THINK_CLOSE_TAGS
|
||||
assert cli_stream_mixin._OPEN_TAGS is THINK_OPEN_TAGS
|
||||
assert cli_stream_mixin._CLOSE_TAGS is THINK_CLOSE_TAGS
|
||||
|
||||
@pytest.mark.parametrize("name", THINK_TAG_NAMES)
|
||||
def test_every_scrubber_tag_is_filtered_when_streamed(self, name):
|
||||
"""A tag added to the shared list is automatically hidden by the gateway stream filter."""
|
||||
c = _make_consumer()
|
||||
for chunk in (f"<{name}>", "hidden", f"</{name}>", "visible"):
|
||||
c._filter_and_accumulate(chunk)
|
||||
c._flush_think_buffer()
|
||||
assert c._accumulated == "visible"
|
||||
|
||||
def test_prose_mention_not_stripped(self):
|
||||
"""<think> mentioned mid-line in prose should NOT trigger filtering."""
|
||||
c = _make_consumer()
|
||||
|
||||
Reference in New Issue
Block a user