ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

merge simp/r3-34-H into simp/r3-34

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:31:56 -07:00
parent bc0ef8d006 04247b9840
commit 394bc29c35
10 changed files with 266 additions and 705 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/spill_safety.py
+14 -35
View File
@@ -1,22 +1,11 @@
"""Symlink-safe creation helpers for spill/cache files.
Spill files (terminal output, hook context, subagent summaries, web_extract
text) live in predictable directories under ``~/.hermes``; a plain
``open(path, "w")`` there would follow a pre-planted symlink and let a local
process redirect the write onto ``~/.bashrc``, ``authorized_keys``, etc.
Every helper refuses symlinks by construction: new files use
``O_CREAT | O_EXCL`` (fails on ANY existing path, including a dangling link);
overwrites ``lstat`` + ``unlink`` the existing path first (removes the link,
never its target) and then create exclusively, so the pair can't be raced.
Privacy tiers: ``private=True`` (default) forces ``0o700`` dirs / ``0o600``
files for spills that may hold pre-redaction secrets; ``private=False`` keeps
umask-default perms for cache dirs bind-mounted into remote terminal backends
(``credential_files._CACHE_DIRS``), where a non-root container UID must read them.
Disk failures are the caller's concern: helpers raise ``OSError``.
"""
"""Symlink-safe creation helpers for spill/cache files under ``~/.hermes``, where a
plain ``open(path, "w")`` would follow a pre-planted symlink onto ``~/.bashrc`` etc.
New files use ``O_CREAT | O_EXCL`` (fails on ANY existing path, even a dangling
link); overwrites ``lstat`` + ``unlink`` first (removes the link, never its target)
then create exclusively, so the pair can't be raced. ``private=True`` (default) =
``0o700`` dirs / ``0o600`` files for spills that may hold pre-redaction secrets;
``private=False`` keeps umask perms for cache dirs bind-mounted into remote backends
(``credential_files._CACHE_DIRS``). Disk failures raise ``OSError`` to the caller."""
from __future__ import annotations
@@ -25,11 +14,7 @@ import stat
from pathlib import Path
from typing import IO
__all__ = [
"ensure_spill_dir",
"open_exclusive",
"write_text_exclusive",
]
__all__ = ["ensure_spill_dir", "open_exclusive", "write_text_exclusive"]
# O_NOFOLLOW is POSIX-only; on Windows O_EXCL alone already refuses every
# pre-existing path.
@@ -38,10 +23,8 @@ _O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
def ensure_spill_dir(path: Path, *, private: bool = True) -> Path:
"""Create ``path`` (and parents) as a directory, refusing symlinks.
``private=True`` creates the leaf ``0o700`` and tightens an existing leaf to
``0o700``. Raises ``OSError`` if the leaf is not a real directory.
"""
``private=True`` creates the leaf ``0o700`` and tightens an existing leaf.
Raises ``OSError`` if the leaf is not a real directory."""
path = Path(path)
path.mkdir(mode=0o700 if private else 0o777, parents=True, exist_ok=True)
st = os.lstat(path)
@@ -58,14 +41,11 @@ def open_exclusive(
private: bool = True,
overwrite: bool = False,
encoding: str = "utf-8",
errors: str = "strict",
) -> IO[str]:
errors: str = "strict") -> IO[str]:
"""Open ``path`` for writing via exclusive create; never follows a link.
``overwrite=True`` first unlinks an existing path (``lstat``-checked, so only
the link itself is removed and directories are refused), then creates
exclusively — the overwrite path cannot be redirected through a symlink either.
"""
exclusively — the overwrite path cannot be redirected through a symlink either."""
path = Path(path)
if overwrite:
try:
@@ -92,8 +72,7 @@ def write_text_exclusive(
private: bool = True,
overwrite: bool = False,
encoding: str = "utf-8",
errors: str = "strict",
) -> None:
errors: str = "strict") -> None:
"""``Path.write_text`` equivalent that refuses to follow symlinks."""
with open_exclusive(
path, private=private, overwrite=overwrite, encoding=encoding, errors=errors
tools/tip_tool.py
+5 -19
View File
@@ -1,13 +1,8 @@
#!/usr/bin/env python3
"""Point at something in the Hermes desktop GUI and say one line about it.
The quiet sibling of ``tour``: same ``data-tour`` handles and discovery call,
but an arrow bubble with no scrim/spotlight/paging. Fire-and-forget — a tip is
not a question, so blocking the turn on a round-trip would stall the reply.
Lives in ``desktop_ui`` (GUI sessions only) and withdraws itself when the user
turns tips off, so the model is never offered a tool whose call would fail.
"""
"""Point at something in the Hermes desktop GUI and say one line about it — the quiet
sibling of ``tour`` (same ``data-tour`` handles) with no scrim/spotlight/paging.
Fire-and-forget: a tip is not a question, so blocking on a round-trip would stall the
reply. Lives in ``desktop_ui`` and withdraws itself when the user turns tips off."""
import json
@@ -21,10 +16,8 @@ def tip_tool(text: str, selector: str, title: str = "", side: str = "") -> str:
"""Show one tip bubble anchored to ``selector``."""
text = (text or "").strip()
selector = (selector or "").strip()
if not text:
return tool_error("tip needs text — the one line the bubble says.")
if not selector:
return tool_error(
"tip needs a selector to point at. Call tour(action='targets') to see "
@@ -33,17 +26,14 @@ def tip_tool(text: str, selector: str, title: str = "", side: str = "") -> str:
if side and side not in SIDES:
return tool_error(f"side must be one of: {', '.join(SIDES)}.")
payload = {"selector": selector, "text": text}
payload.update({k: v for k, v in (("title", title), ("side", side)) if v})
try:
ok = desktop_ui.emit("tip.show", payload)
except Exception as exc:
return tool_error(f"Failed to show the tip: {exc}")
if not ok:
return tool_error("tip is only available in the Hermes desktop app.")
return json.dumps({"success": True, "selector": selector}, ensure_ascii=False)
@@ -93,11 +83,7 @@ registry.register(
toolset="desktop_ui",
schema=TIP_SCHEMA,
handler=lambda args, **kw: tip_tool(
text=args.get("text", ""),
selector=args.get("selector", ""),
title=args.get("title", ""),
side=args.get("side", ""),
),
**{k: args.get(k, "") for k in ("text", "selector", "title", "side")}),
check_fn=check_tips_enabled,
emoji="💡",
)
tools/todo_tool.py
+17 -44
View File
@@ -27,13 +27,13 @@ _TRUNCATION_MARKER = "… [truncated]"
TODO_INJECTION_HEADER = (
"[Your active task list was preserved across context compression]"
)
_STATUS_MARKERS = {"completed": "[x]", "in_progress": "[>]", "pending": "[ ]", "cancelled": "[~]"}
_ACTIVE_STATUSES = {"pending", "in_progress"}
class TodoStore:
"""In-memory todo list, one per AIAgent. List position is priority.
Items: ``{id, content, status, parent?}`` — ``parent`` nests a subtask.
"""
Items: ``{id, content, status, parent?}`` — ``parent`` nests a subtask."""
def __init__(self):
self._items: List[Dict[str, str]] = []
@@ -46,13 +46,12 @@ class TodoStore:
def write(self, todos: List[Dict[str, Any]], merge: bool = False) -> List[Dict[str, str]]:
"""Replace the list (default) or merge by id; returns the full list after writing."""
before = self.read()
if not merge:
self._items = self._fresh_items(todos)
else:
if merge:
self._merge(todos)
else:
self._items = self._fresh_items(todos)
# Keep the highest-priority head so a replayed list can't grow re-injection unbounded.
if len(self._items) > MAX_TODO_ITEMS:
self._items = self._items[:MAX_TODO_ITEMS]
del self._items[MAX_TODO_ITEMS:]
self._sanitize_parents(self._items)
if self._items != before:
self._revision += 1
@@ -104,12 +103,7 @@ class TodoStore:
"""Return the full state clients can reconcile atomically."""
return {"todos": self.read(), "revision": self._revision}
def restore(
self,
todos: List[Dict[str, Any]],
*,
revision: Any = 0,
) -> List[Dict[str, str]]:
def restore(self, todos: List[Dict[str, Any]], *, revision: Any = 0) -> List[Dict[str, str]]:
"""Restore a trusted snapshot without manufacturing a new revision."""
self._items = self._fresh_items(todos)[:MAX_TODO_ITEMS]
try:
@@ -120,16 +114,11 @@ class TodoStore:
def format_for_injection(self) -> Optional[str]:
"""Render the list for post-compression injection, or None if nothing active.
Only pending/in_progress items are injected — finished ones make the
model re-do work after compression. A parent is kept (with its real
status marker) when any descendant is active so subtasks keep context.
"""
status marker) when any descendant is active so subtasks keep context."""
if not self._items:
return None
markers = {"completed": "[x]", "in_progress": "[>]", "pending": "[ ]", "cancelled": "[~]"}
active = {"pending", "in_progress"}
children: Dict[str, List[Dict[str, str]]] = {}
roots: List[Dict[str, str]] = []
for item in self._items:
@@ -144,9 +133,9 @@ class TodoStore:
has_active_kid = False
for kid in children.get(item["id"], []):
has_active_kid |= render(kid, depth + 1, kid_lines)
keep = item["status"] in active or has_active_kid
keep = item["status"] in _ACTIVE_STATUSES or has_active_kid
if keep:
marker = markers.get(item["status"], "[?]")
marker = _STATUS_MARKERS.get(item["status"], "[?]")
out.append(
f"{' ' * depth}- {marker} {item['id']}. "
f"{item['content']} ({item['status']})"
@@ -157,17 +146,13 @@ class TodoStore:
lines = [TODO_INJECTION_HEADER]
for item in roots:
render(item, 0, lines)
if len(lines) == 1:
return None
return "\n".join(lines)
return "\n".join(lines) if len(lines) > 1 else None
@staticmethod
def _cap_content(content: str) -> str:
"""Truncate to MAX_TODO_CONTENT_CHARS keeping the head (the actionable part) + marker."""
if len(content) > MAX_TODO_CONTENT_CHARS:
keep = MAX_TODO_CONTENT_CHARS - len(_TRUNCATION_MARKER)
return content[:keep] + _TRUNCATION_MARKER
return content[:MAX_TODO_CONTENT_CHARS - len(_TRUNCATION_MARKER)] + _TRUNCATION_MARKER
return content
@staticmethod
@@ -175,14 +160,12 @@ class TodoStore:
"""Normalize one item to ``{id, content, status, parent?}`` with placeholders for missing fields."""
if not isinstance(item, dict):
return {"id": "?", "content": "(invalid item)", "status": "pending"}
item_id = str(item.get("id", "")).strip() or "?"
content = str(item.get("content", "")).strip()
content = TodoStore._cap_content(content) if content else "(no description)"
status = str(item.get("status", "pending")).strip().lower()
if status not in VALID_STATUSES:
status = "pending"
result = {"id": item_id, "content": content, "status": status}
parent = str(item.get("parent") or "").strip()
if parent and parent != item_id:
@@ -215,16 +198,13 @@ class TodoStore:
# Non-dict items get a synthetic key so _validate can handle them
last_index[f"__invalid_{i}"] = i
continue
item_id = str(item.get("id", "")).strip() or "?"
last_index[item_id] = i
last_index[str(item.get("id", "")).strip() or "?"] = i
return [todos[i] for i in sorted(last_index.values())]
@staticmethod
def _normalize_order(items: List[Dict[str, str]]) -> List[Dict[str, str]]:
"""Lift the in_progress step ahead of any earlier pending placeholder.
Nested lists keep authored order — reordering would tear a subtask from its siblings.
"""
Nested lists keep authored order — reordering would tear a subtask from its siblings."""
if any(item.get("parent") for item in items):
return items
statuses = [item["status"] for item in items]
@@ -233,11 +213,8 @@ class TodoStore:
active_index = statuses.index("in_progress")
if "pending" not in statuses[:active_index]:
return items
pending_index = statuses.index("pending")
normalized = items.copy()
active_item = normalized.pop(active_index)
normalized.insert(pending_index, active_item)
normalized.insert(statuses.index("pending"), normalized.pop(active_index))
return normalized
@@ -249,7 +226,6 @@ def todo_tool(
"""Write ``todos`` (replace or ``merge`` by id) or read when None; returns list + summary JSON."""
if store is None:
return tool_error("TodoStore not initialized")
if todos is not None:
if isinstance(todos, str): # LLMs sometimes send a JSON string instead of a list
try:
@@ -257,9 +233,7 @@ def todo_tool(
except (json.JSONDecodeError, TypeError):
return tool_error("todos must be a list of objects, got unparseable string")
if not isinstance(todos, list):
return tool_error(
f"todos must be a list, got {type(todos).__name__}"
)
return tool_error(f"todos must be a list, got {type(todos).__name__}")
items = store.write(todos, merge)
else:
items = store.read()
@@ -267,7 +241,6 @@ def todo_tool(
summary = {"total": len(items)}
for status in ("pending", "in_progress", "completed", "cancelled"):
summary[status] = sum(1 for i in items if i["status"] == status)
return json.dumps({
"todos": items,
"revision": store.snapshot()["revision"],
tools/tool_backend_helpers.py
+39 -96
View File
@@ -18,30 +18,21 @@ _VALID_MODAL_MODES = {"auto", "direct", "managed"}
def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
"""True when the user is entitled to the Nous Tool Gateway (coarse gate:
paid Portal service access OR a live free tool pool — "is ANY managed tool usable").
Fails closed on unknown/error entitlement — never blocks startup.
Per-category coverage is narrowed by callers via ``tool_gateway_entitled_for``.
``force_fresh=True`` is for interactive flows that must see a just-purchased grant.
"""
"""Coarse gate: entitled to the Nous Tool Gateway (paid Portal access OR a live free
pool). Fails closed on unknown/error — never blocks startup. Callers narrow per category
via ``tool_gateway_entitled_for``; ``force_fresh`` is for flows needing a just-bought grant."""
try:
from hermes_cli.nous_account import get_nous_portal_account_info
if force_fresh:
account_info = get_nous_portal_account_info(force_fresh=True)
else:
account_info = get_nous_portal_account_info()
account_info = (get_nous_portal_account_info(force_fresh=True) if force_fresh
else get_nous_portal_account_info())
return bool(account_info.logged_in) and account_info.tool_gateway_entitled
except Exception:
return False
def nous_tool_gateway_unavailable_message(
capability: str = "the Nous Tool Gateway",
*,
force_fresh: bool = False,
) -> str:
capability: str = "the Nous Tool Gateway", *, force_fresh: bool = False) -> str:
"""Return account-aware guidance for an unavailable Nous Tool Gateway path."""
try:
from hermes_cli.nous_account import (
@@ -50,18 +41,14 @@ def nous_tool_gateway_unavailable_message(
)
account_info = get_nous_portal_account_info(force_fresh=force_fresh)
message = format_nous_portal_entitlement_message(
account_info,
capability=capability,
)
message = format_nous_portal_entitlement_message(account_info, capability=capability)
if message:
return message
except Exception:
pass
return (
f"{capability} is unavailable. Run `hermes model` to refresh your "
"Nous Portal login and billing status."
)
"Nous Portal login and billing status.")
def normalize_browser_cloud_provider(value: object | None) -> str:
@@ -76,9 +63,7 @@ def coerce_modal_mode(value: object | None) -> str:
return mode if mode in _VALID_MODAL_MODES else _DEFAULT_MODAL_MODE
def normalize_modal_mode(value: object | None) -> str:
"""Return a normalized modal execution mode."""
return coerce_modal_mode(value)
normalize_modal_mode = coerce_modal_mode
def has_direct_modal_credentials() -> bool:
@@ -88,9 +73,7 @@ def has_direct_modal_credentials() -> bool:
except (PermissionError, OSError):
modal_file_exists = False
return bool(
(os.getenv("MODAL_TOKEN_ID") and os.getenv("MODAL_TOKEN_SECRET"))
or modal_file_exists
)
(os.getenv("MODAL_TOKEN_ID") and os.getenv("MODAL_TOKEN_SECRET")) or modal_file_exists)
def resolve_modal_backend_state(
@@ -98,15 +81,12 @@ def resolve_modal_backend_state(
*,
has_direct: bool,
managed_ready: bool,
managed_enabled: bool | None = None,
) -> Dict[str, Any]:
managed_enabled: bool | None = None) -> Dict[str, Any]:
"""Resolve direct vs managed Modal backend: ``direct``/``managed`` are
exclusive; ``auto`` prefers managed when available, else direct."""
requested_mode = coerce_modal_mode(modal_mode)
if managed_enabled is None:
managed_enabled = managed_nous_tools_enabled()
managed_mode_blocked = requested_mode == "managed" and not managed_enabled
managed_ok = managed_enabled and managed_ready
if requested_mode == "managed":
selected_backend = "managed" if managed_ok else None
@@ -114,15 +94,13 @@ def resolve_modal_backend_state(
selected_backend = "direct" if has_direct else None
else:
selected_backend = "managed" if managed_ok else "direct" if has_direct else None
return {
"requested_mode": requested_mode,
"mode": requested_mode,
"has_direct": has_direct,
"managed_ready": managed_ready,
"managed_mode_blocked": managed_mode_blocked,
"selected_backend": selected_backend,
}
"managed_mode_blocked": requested_mode == "managed" and not managed_enabled,
"selected_backend": selected_backend}
def _scoped_credential(name: str) -> str:
@@ -137,12 +115,18 @@ def _scoped_credential(name: str) -> str:
return (os.getenv(name, "") or "").strip()
def _dotenv_value(env_var: str) -> str:
"""``.env`` value via ``hermes_cli.config.get_env_value`` (``""`` when unavailable)."""
try:
from hermes_cli.config import get_env_value
return str(get_env_value(env_var) or "").strip()
except Exception: # pragma: no cover — config is in-repo
return ""
def resolve_provider_secret(
env_var: str,
provider_id: str,
config_value: str = "",
env_getter=None,
) -> str:
env_var: str, provider_id: str, config_value: str = "", env_getter=None) -> str:
"""Resolve a voice-provider API key (single owner for STT/TTS lookup).
Order: explicit ``config_value`` -> profile secret scope / env -> ``.env``
@@ -154,11 +138,9 @@ def resolve_provider_secret(
value = str(config_value or "").strip()
if value:
return value
key = _scoped_credential(env_var)
if key:
return key
try:
from agent.secret_scope import is_multiplex_active
@@ -166,21 +148,10 @@ def resolve_provider_secret(
return ""
except Exception: # pragma: no cover — secret_scope is in-repo
pass
if env_getter is not None:
key = str(env_getter(env_var) or "").strip()
else:
try:
from hermes_cli.config import get_env_value
key = str(get_env_value(env_var) or "").strip()
except ImportError: # pragma: no cover — config is in-repo
key = ""
if key:
key = (str(env_getter(env_var) or "").strip() if env_getter is not None
else _dotenv_value(env_var))
if key or not provider_id:
return key
if not provider_id:
return ""
try:
from agent.credential_pool import load_pool
@@ -193,9 +164,7 @@ def resolve_provider_secret(
if entry is None:
continue
key = str(
getattr(entry, "runtime_api_key", "")
or getattr(entry, "access_token", "")
or ""
getattr(entry, "runtime_api_key", "") or getattr(entry, "access_token", "") or ""
).strip()
if key:
return key
@@ -207,19 +176,14 @@ def resolve_provider_secret(
def resolve_openai_audio_api_key() -> str:
"""Prefer VOICE_TOOLS_OPENAI_KEY, else OPENAI_API_KEY (scope-aware, with
credential-pool fallback for the latter). Must go through the secret scope:
a raw ``os.environ`` read could bill another profile's account under multiplex.
"""
a raw ``os.environ`` read could bill another profile's account under multiplex."""
return (
resolve_provider_secret("VOICE_TOOLS_OPENAI_KEY", "")
or resolve_provider_secret("OPENAI_API_KEY", "openai-api")
)
or resolve_provider_secret("OPENAI_API_KEY", "openai-api"))
def prefers_gateway(config_section: str) -> bool:
"""Return True when the user opted into the Tool Gateway for this tool.
Reads ``<section>.use_gateway`` from config.yaml. Never raises.
"""
"""True when ``<section>.use_gateway`` is set in config.yaml. Never raises."""
try:
from hermes_cli.config import load_config
section = (load_config() or {}).get(config_section)
@@ -235,16 +199,11 @@ def prefers_gateway(config_section: str) -> bool:
NOUS_MANAGED_PROVIDER = "nous"
# Per-capability keys that also count as "this category has been configured".
_EXTRA_SELECTION_KEYS = {
"web": ("search_backend", "extract_backend"),
}
_EXTRA_SELECTION_KEYS = {"web": ("search_backend", "extract_backend")}
# Key(s) carrying the category's provider selection. ``browser.backend`` is the
# DRIVER choice (browser-use CLI vs built-in), not the cloud provider — excluded.
_SELECTION_NAME_KEYS = {
"browser": ("cloud_provider",),
"web": ("backend",),
}
_SELECTION_NAME_KEYS = {"browser": ("cloud_provider",), "web": ("backend",)}
_DEFAULT_NAME_KEYS = ("provider", "backend", "cloud_provider")
@@ -273,10 +232,8 @@ def read_selection(section: str) -> str | None:
raw = _raw_section(section)
if raw is None:
return None
if "use_gateway" in raw and is_truthy_value(raw.get("use_gateway"), default=False):
return NOUS_MANAGED_PROVIDER
for key in _SELECTION_NAME_KEYS.get(section, _DEFAULT_NAME_KEYS):
value = raw.get(key)
if value is not None:
@@ -302,43 +259,29 @@ def selection_exists(section: str) -> bool:
# Backends that once shipped in-tree but were removed. A config still pointing at
# one would otherwise fail silently at the FIRST tool call with a generic "no
# registered provider has that name" — no migration, no startup notice. Consulted
# by the startup config check and selection_error() so the user gets a real message.
# Add removals here, never as one-off string checks at call sites, e.g.
# registered provider has that name". Consulted by the startup config check and
# selection_error(). Add removals here, never as one-off string checks, e.g.
# "web": {"<name>": "the <Name> backend was removed in vX.Y.Z (...)"},
REMOVED_BACKENDS: Dict[str, Dict[str, str]] = {}
def removed_backend_note(section: str, name: str) -> Optional[str]:
"""Explanation for a backend that used to ship in-tree, or None.
``name`` tolerates the quoted form callers pass to selection_error().
"""
``name`` tolerates the quoted form callers pass to selection_error()."""
normalized = (name or "").strip().strip("'\"").lower()
return REMOVED_BACKENDS.get(section, {}).get(normalized)
def selection_error(section: str, selection_name: str, failure: str) -> str:
"""The uniform honest-error contract for a selected-but-broken provider."""
note = removed_backend_note(section, selection_name)
if note:
failure = note
failure = removed_backend_note(section, selection_name) or failure
return (
f"{section} is configured to use {selection_name} (set via hermes "
f"tools), but {failure}. Run 'hermes tools' to change it."
)
f"tools), but {failure}. Run 'hermes tools' to change it.")
def fal_key_is_configured() -> bool:
"""True when FAL_KEY is set (scope/env, else ``.env`` for CLI paths that
run before dotenv loads) to a non-whitespace value — so tool-side and CLI
setup-time checks agree; whitespace-only counts as unset everywhere."""
value = _scoped_credential("FAL_KEY") or None
if value is None:
try:
from hermes_cli.config import get_env_value
value = get_env_value("FAL_KEY")
except Exception:
value = None
return bool(value and value.strip())
return bool(_scoped_credential("FAL_KEY") or _dotenv_value("FAL_KEY"))
tools/tool_output_limits.py
+8 -21
View File
@@ -1,16 +1,8 @@
"""Configurable tool-output truncation limits (``tool_output`` in config.yaml).
Centralises the caps previously hardcoded in ``terminal_tool`` (``max_bytes``)
and ``file_operations`` (``max_lines`` / ``max_line_length``). Defaults equal
the old constants, so behaviour is unchanged when the section is absent, and
the reader never raises — any config error falls back to the defaults.
Ported from anomalyco/opencode PR #23770. Example::
tool_output:
max_bytes: 100000 # terminal output cap (chars)
max_lines: 5000 # read_file pagination + truncation cap
max_line_length: 2000 # per-line cap before '... [truncated]'
"""
"""Configurable tool-output truncation limits (``tool_output`` in config.yaml):
``max_bytes`` (terminal output cap), ``max_lines`` (read_file pagination cap),
``max_line_length`` (per-line cap before '... [truncated]'). Defaults equal the
constants once hardcoded in terminal_tool / file_operations and the reader never
raises, so behaviour is unchanged when the section is absent or malformed."""
from __future__ import annotations
@@ -39,11 +31,8 @@ def _coerce_positive_int(value: Any, default: int) -> int:
def get_tool_output_limits() -> Dict[str, int]:
"""Return resolved limits ``{max_bytes, max_lines, max_line_length}``; never raises.
Cached for the process lifetime — ``_reset_tool_output_limits_cache()``
forces a fresh read after config changes.
"""
"""Resolved ``{max_bytes, max_lines, max_line_length}``; never raises. Cached
for the process — ``_reset_tool_output_limits_cache()`` forces a fresh read."""
global _cached_limits
if _cached_limits is not None:
return _cached_limits
@@ -55,13 +44,11 @@ def get_tool_output_limits() -> Dict[str, int]:
section = {}
except Exception:
section = {}
_cached_limits = {
"max_bytes": _coerce_positive_int(section.get("max_bytes"), DEFAULT_MAX_BYTES),
"max_lines": _coerce_positive_int(section.get("max_lines"), DEFAULT_MAX_LINES),
"max_line_length": _coerce_positive_int(
section.get("max_line_length"), DEFAULT_MAX_LINE_LENGTH
),
section.get("max_line_length"), DEFAULT_MAX_LINE_LENGTH),
}
return _cached_limits
tools/tool_result_storage.py
+61 -139
View File
@@ -1,24 +1,11 @@
"""Tool result persistence -- preserves large outputs instead of truncating.
Three layers of defense against context-window overflow:
1. **Per-tool output cap** (inside each tool, e.g. search_files pre-truncates).
2. **Per-result persistence** (``maybe_persist_tool_result``): output over the
tool's registered threshold is persisted and replaced in-context by a
preview + file path. The canonical home is ALWAYS host-side,
``$HERMES_HOME/cache/spillover/{tool_use_id}.txt``, so it works for sessions
that never ran a terminal (MCP-only, cron, gateway). Local backend / no env:
the model sees the host path. Remote backends (docker/ssh/modal/daytona):
``cache/spillover`` is in the auto-mounted/synced cache-dir list
(tools/credential_files.py), so the model sees the translated in-sandbox
path — probed for readability first, falling back to a copy written into
the sandbox temp dir (e.g. a persistent container created before spillover
joined the mount list). The dir is pruned hourly by gateway housekeeping
and once per process on first spill (CLI-only installs never run housekeeping).
3. **Per-turn aggregate budget** (``enforce_turn_budget``): if all tool results
in one turn exceed the turn budget, the largest non-persisted results are
spilled until under budget.
"""
Layers against context overflow: (1) per-tool caps inside each tool; (2)
``maybe_persist_tool_result`` — output over the tool's threshold is persisted and
replaced by a preview + path. Canonical home is ALWAYS host-side
``$HERMES_HOME/cache/spillover/{id}.txt`` (works for sessions that never ran a
terminal); remote backends get the translated in-sandbox path (probed for
readability) else a copy in the sandbox temp dir. (3) ``enforce_turn_budget``."""
import hashlib
import logging
@@ -28,11 +15,7 @@ import shlex
import threading
import time
from tools.budget_config import (
DEFAULT_PREVIEW_SIZE_CHARS,
BudgetConfig,
DEFAULT_BUDGET,
)
from tools.budget_config import DEFAULT_PREVIEW_SIZE_CHARS, BudgetConfig, DEFAULT_BUDGET
logger = logging.getLogger(__name__)
PERSISTED_OUTPUT_TAG = "<persisted-output>"
@@ -57,10 +40,7 @@ def get_spillover_dir():
def cleanup_spillover_cache(max_age_hours: int = SPILLOVER_MAX_AGE_HOURS) -> int:
"""Delete spillover files older than *max_age_hours*; returns count removed.
Same contract as the ``cleanup_*_cache`` helpers in ``gateway.platforms.base``
so the gateway housekeeping loop can prune this dir on the hourly cadence.
"""
Same contract as the ``cleanup_*_cache`` helpers the gateway housekeeping loop prunes hourly."""
cutoff = time.time() - (max_age_hours * 3600)
removed = 0
try:
@@ -93,11 +73,9 @@ def _prune_spillover_once() -> None:
def _is_host_side_env(env) -> bool:
"""True when this process should write the spill file directly.
``env=None`` (no sandbox yet) and the local backend qualify; remote backends
resolve ``read_file`` inside the sandbox, so the spill must be written there.
"""
"""True when this process should write the spill file directly: ``env=None``
(no sandbox yet) or the local backend. Remote backends resolve ``read_file``
inside the sandbox, so the spill must be written there."""
if env is None:
return True
try:
@@ -123,13 +101,10 @@ def _write_to_spillover(content: str, filename: str):
def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
"""Return the path where a remote backend can read *host_path*, or None.
Translates via the same helper the image tools use, forces a sync for synced
backends, then PROBES readability — a persistent docker container created
before spillover joined the mount list lacks the bind mount and must fall
back to the in-sandbox write.
"""
"""Path where a remote backend can read *host_path*, or None. Translates via
the image tools' helper, forces a sync for synced backends, then PROBES
readability — a persistent container created before spillover joined the
mount list lacks the bind mount and must fall back to the in-sandbox write."""
try:
from tools.credential_files import to_agent_visible_cache_path
@@ -137,14 +112,12 @@ def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
except Exception as exc:
logger.debug("Spillover path translation failed: %s", exc)
return None
sync_manager = getattr(env, "_sync_manager", None)
if sync_manager is not None:
try:
sync_manager.sync(force=True)
except Exception as exc:
logger.debug("Spillover sync failed: %s", exc)
try:
result = env.execute(f"test -r {shlex.quote(visible)}", timeout=15)
if result.get("returncode", 1) == 0:
@@ -156,17 +129,15 @@ def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
def _resolve_storage_dir(env) -> str:
"""Return the best temp-backed storage dir for this environment."""
if env is not None:
get_temp_dir = getattr(env, "get_temp_dir", None)
if callable(get_temp_dir):
try:
temp_dir = get_temp_dir()
except Exception as exc:
logger.debug("Could not resolve env temp dir: %s", exc)
else:
if temp_dir:
temp_dir = temp_dir.rstrip("/") or "/"
return f"{temp_dir}/hermes-results"
get_temp_dir = getattr(env, "get_temp_dir", None)
if callable(get_temp_dir):
try:
temp_dir = get_temp_dir()
except Exception as exc:
logger.debug("Could not resolve env temp dir: %s", exc)
else:
if temp_dir:
return f"{temp_dir.rstrip('/') or '/'}/hermes-results"
return STORAGE_DIR
@@ -175,16 +146,13 @@ def _safe_result_filename(tool_use_id: str) -> str:
raw_id = str(tool_use_id or "tool_result")
safe_stem = _UNSAFE_RESULT_FILENAME_CHARS.sub("_", raw_id).strip("._-")
changed = safe_stem != raw_id
if not safe_stem:
safe_stem = "tool_result"
changed = True
if changed or len(safe_stem) > _MAX_RESULT_FILENAME_STEM:
digest = hashlib.sha256(raw_id.encode("utf-8")).hexdigest()[:12]
safe_stem = safe_stem[:_MAX_RESULT_FILENAME_STEM].rstrip("._-") or "tool_result"
safe_stem = f"{safe_stem}_{digest}"
return f"{safe_stem}.txt"
@@ -201,11 +169,9 @@ def generate_preview(content: str, max_chars: int = DEFAULT_PREVIEW_SIZE_CHARS)
def _write_to_sandbox(content: str, remote_path: str, env) -> bool:
"""Write content into the sandbox via env.execute(). Returns True on success.
Content goes through stdin, not the command string: Linux ``MAX_ARG_STRLEN``
caps a single argv element at 128 KB, so a heredoc-in-command silently failed
for exactly the oversized results persistence exists to handle.
"""
caps one argv element at 128 KB, so a heredoc-in-command silently failed
for exactly the oversized results persistence exists to handle."""
storage_dir = os.path.dirname(remote_path)
cmd = f"mkdir -p {shlex.quote(storage_dir)} && cat > {shlex.quote(remote_path)}"
result = env.execute(cmd, timeout=30, stdin_data=content)
@@ -213,41 +179,30 @@ def _write_to_sandbox(content: str, remote_path: str, env) -> bool:
def _build_persisted_message(
preview: str,
has_more: bool,
original_size: int,
file_path: str,
) -> str:
preview: str, has_more: bool, original_size: int, file_path: str) -> str:
"""Build the <persisted-output> replacement block."""
size_kb = original_size / 1024
size_str = f"{size_kb / 1024:.1f} MB" if size_kb >= 1024 else f"{size_kb:.1f} KB"
msg = f"{PERSISTED_OUTPUT_TAG}\n"
msg += f"This tool result was too large ({original_size:,} characters, {size_str}).\n"
msg += f"Full output saved to: {file_path}\n"
msg += "Use the read_file tool with offset and limit to access specific sections of this output.\n"
msg += (
return (
f"{PERSISTED_OUTPUT_TAG}\n"
f"This tool result was too large ({original_size:,} characters, {size_str}).\n"
f"Full output saved to: {file_path}\n"
"Use the read_file tool with offset and limit to access specific sections of this output.\n"
"Recovery: page through the saved file with read_file (offset/limit) or "
"process it with execute_code — do NOT re-request the same data from the "
"remote API; the full result is already on disk.\n\n"
)
msg += f"Preview (first {len(preview)} chars):\n"
msg += preview
if has_more:
msg += "\n..."
msg += f"\n{PERSISTED_OUTPUT_CLOSING_TAG}"
return msg
f"Preview (first {len(preview)} chars):\n"
+ preview + ("\n..." if has_more else "")
+ f"\n{PERSISTED_OUTPUT_CLOSING_TAG}")
_PERSISTED_PATH_RE = re.compile(r"^Full output saved to: (.+)$", re.MULTILINE)
def extract_persisted_path(content: str) -> str | None:
"""Return the file path from a <persisted-output> block, or None.
Lets the result-reference stubbing guard (agent/tool_guardrails.py) carry
the spillover path in a stub instead of leaving it dangling.
"""
"""Return the file path from a <persisted-output> block, or None. Lets the
result-reference stubbing guard (agent/tool_guardrails.py) carry the
spillover path in a stub instead of leaving it dangling."""
if not isinstance(content, str) or PERSISTED_OUTPUT_TAG not in content:
return None
match = _PERSISTED_PATH_RE.search(content)
@@ -260,19 +215,12 @@ def maybe_persist_tool_result(
tool_use_id: str,
env=None,
config: BudgetConfig = DEFAULT_BUDGET,
threshold: int | float | None = None,
) -> str:
threshold: int | float | None = None) -> str:
"""Layer 2: persist an oversized result, return preview + path.
``threshold`` overrides ``config.resolve_threshold(tool_name)``. Falls back
to inline truncation when no write location succeeds.
"""
to inline truncation when no write location succeeds."""
effective_threshold = threshold if threshold is not None else config.resolve_threshold(tool_name)
if effective_threshold == float("inf"):
return content
if len(content) <= effective_threshold:
if effective_threshold == float("inf") or len(content) <= effective_threshold:
return content
filename = _safe_result_filename(tool_use_id)
@@ -281,26 +229,22 @@ def maybe_persist_tool_result(
def _persisted(path: str, host_suffix: str = "") -> str:
logger.info(
"Persisted large tool result: %s (%s, %d chars -> %s%s)",
tool_name, tool_use_id, len(content), path, host_suffix,
)
tool_name, tool_use_id, len(content), path, host_suffix)
return _build_persisted_message(preview, has_more, len(content), path)
# Always persist host-side first: cache/spillover is the single canonical
# home for spilled results regardless of backend.
# Always persist host-side first: cache/spillover is the single canonical home.
host_path = _write_to_spillover(content, filename)
if _is_host_side_env(env):
if host_path is not None:
return _persisted(host_path)
elif env is not None:
# Remote backend: reference the mounted/synced path when the sandbox
# can actually read it ...
else:
# Remote backend: reference the mounted/synced path when the sandbox can
# actually read it, else write into the sandbox temp dir (pre-existing
# containers without the spillover mount, translation/probe failures).
if host_path is not None:
visible = _sandbox_visible_spillover_path(host_path, env)
if visible is not None:
return _persisted(visible, f" [host: {host_path}]")
# ... else write into the sandbox temp dir (pre-existing containers
# without the spillover mount, translation/probe failures).
remote_path = f"{_resolve_storage_dir(env)}/{filename}"
try:
if _write_to_sandbox(content, remote_path, env):
@@ -310,61 +254,39 @@ def maybe_persist_tool_result(
logger.info(
"Inline-truncating large tool result: %s (%d chars, no sandbox write)",
tool_name, len(content),
)
tool_name, len(content))
return (
f"{preview}\n\n"
f"[Truncated: tool response was {len(content):,} chars. "
f"Full output could not be saved to sandbox.]"
)
f"Full output could not be saved to sandbox.]")
def enforce_turn_budget(
tool_messages: list[dict],
env=None,
config: BudgetConfig = DEFAULT_BUDGET,
) -> list[dict]:
"""Layer 3: enforce the aggregate budget across all tool results in a turn.
Persists the largest non-persisted results first until under budget.
Mutates the list in-place and returns it.
"""
tool_messages: list[dict], env=None, config: BudgetConfig = DEFAULT_BUDGET) -> list[dict]:
"""Layer 3: persist the largest non-persisted results first until the turn's
aggregate is under budget. Mutates the list in-place and returns it."""
candidates = []
total_size = 0
for i, msg in enumerate(tool_messages):
content = msg.get("content", "")
size = len(content)
size = len(msg.get("content", ""))
total_size += size
if PERSISTED_OUTPUT_TAG not in content:
if PERSISTED_OUTPUT_TAG not in msg.get("content", ""):
candidates.append((i, size))
if total_size <= config.turn_budget:
return tool_messages
candidates.sort(key=lambda x: x[1], reverse=True)
for idx, size in candidates:
if total_size <= config.turn_budget:
break
msg = tool_messages[idx]
content = msg["content"]
tool_use_id = msg.get("tool_call_id", f"budget_{idx}")
content = tool_messages[idx]["content"]
tool_use_id = tool_messages[idx].get("tool_call_id", f"budget_{idx}")
replacement = maybe_persist_tool_result(
content=content,
tool_name=_BUDGET_TOOL_NAME,
tool_use_id=tool_use_id,
env=env,
config=config,
threshold=0,
)
content=content, tool_name=_BUDGET_TOOL_NAME, tool_use_id=tool_use_id,
env=env, config=config, threshold=0)
if replacement != content:
total_size -= size
total_size += len(replacement)
total_size += len(replacement) - size
tool_messages[idx]["content"] = replacement
logger.info(
"Budget enforcement: persisted tool result %s (%d chars)",
tool_use_id, size,
)
logger.info("Budget enforcement: persisted tool result %s (%d chars)",
tool_use_id, size)
return tool_messages
tools/tool_search.py
+85 -241
View File
@@ -1,20 +1,11 @@
"""Progressive tool disclosure ("tool search"): MCP/plugin tools (and a
curated set of event-triggered core tools) are replaced in the model-visible
array by three bridge tools — tool_search / tool_describe / tool_call.
Invariants:
* Working-set core tools (``toolsets._HERMES_CORE_TOOLS``) and session-gated GUI
toolsets never defer unless named in the ``defer`` list.
* Tiered disclosure: ANY deferrable tool activates the bridge (tier 1 = bridge +
listing that fits ``min(threshold_pct% of context, listing_max_tokens)``,
tier 2 = bare bridge + per-server summary). The listing scales, not activation.
* The catalog is stateless — rebuilt from the live tool-defs list on every
assembly. A session-keyed catalog drifts from the registry and silently
drops tools.
* Bridge calls route through ``model_tools.handle_function_call`` so guardrails,
hooks, approvals, and truncation fire identically; display/trajectory unwrap
always shows the underlying tool.
"""
"""Progressive tool disclosure ("tool search"): MCP/plugin tools and a curated set
of event-triggered core tools are replaced in the model-visible array by three
bridge tools — tool_search / tool_describe / tool_call. Invariants: core tools
(``toolsets._HERMES_CORE_TOOLS``) and session-gated GUI toolsets never defer unless
named in ``defer``; ANY deferrable tool activates the bridge (the listing scales
with budget, not activation); the catalog is stateless — rebuilt from the live
tool-defs every assembly (a session-keyed one drifts and silently drops tools);
bridge calls route through ``model_tools.handle_function_call`` (same guardrails)."""
from __future__ import annotations
@@ -23,52 +14,26 @@ import logging
import math
from collections import Counter
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple
from typing import Any, Dict, Iterable, List, Optional, Tuple
from tools.registry import tool_error
from tools.tool_search_names import ( # noqa: F401 — re-exported public names
BRIDGE_TOOL_NAMES,
TOOL_CALL_NAME,
TOOL_DESCRIBE_NAME,
TOOL_SEARCH_NAME,
BRIDGE_TOOL_NAMES, TOOL_CALL_NAME, TOOL_DESCRIBE_NAME, TOOL_SEARCH_NAME,
)
from tools.tool_search_catalog import ( # noqa: F401 — re-exported public/test names
CHARS_PER_TOKEN,
CatalogEntry,
_classify_source,
_corpus_stats,
_entry_search_text,
_fn,
_listing_group_label,
_registry_entry,
_short_desc,
_stem,
_tokenize,
build_catalog,
build_catalog_listing,
build_catalog_listing_with_form,
search_catalog,
)
from tools.tool_search_validation import ( # noqa: F401 — re-exported public/test names
_schema_for_local_validation,
_schema_has_external_ref,
validate_deferred_call_args,
CHARS_PER_TOKEN, CatalogEntry, _corpus_stats, _entry_search_text, _fn,
_listing_group_label, _registry_entry, _short_desc, _stem, _tokenize,
build_catalog, build_catalog_listing_with_form, search_catalog,
)
from tools.tool_search_validation import validate_deferred_call_args # noqa: F401
logger = logging.getLogger("tools.tool_search")
# Bound the work one tool_search bridge call can request.
# Bound the work one bridge call can request.
_MAX_QUERIES_PER_CALL = 10
# Bound the work one tool_describe bridge call can request.
_MAX_DESCRIBE_NAMES_PER_CALL = 10
# ---------------------------------------------------------------------------
# Configuration plumbing
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class ToolSearchConfig:
"""Resolved, validated tool-search configuration for a single assembly."""
@@ -79,9 +44,8 @@ class ToolSearchConfig:
threshold_pct: float # 0..100
search_default_limit: int
max_search_limit: int
# Embedded name + short-description manifest so deferred tools stay
# DISCOVERABLE (like the skills listing) while full schemas stay deferred.
# "auto"/"on" = include when it fits (names-only, then bare); "off" = bare bridge.
# Embedded name + short-description manifest keeping deferred tools
# DISCOVERABLE. "auto"/"on" = include when it fits; "off" = bare bridge.
listing: str = "auto" # "auto" | "on" | "off"
# Effective budget = min(listing_max_tokens, threshold_pct% of context).
listing_max_tokens: int = 4000
@@ -95,13 +59,11 @@ class ToolSearchConfig:
@classmethod
def from_raw(cls, raw: Any) -> "ToolSearchConfig":
"""Build a config from a raw dict / legacy bool / None. Every field is
validated and clamped; unknown values fall back to safe defaults rather
than raising, so a typo in user config cannot break the agent."""
"""Build from a raw dict / legacy bool / None. Every field is clamped and
unknown values fall back to safe defaults — a config typo must not break the agent."""
if not isinstance(raw, dict):
return cls(enabled="off" if raw is False else "auto", threshold_pct=5.0,
search_default_limit=5, max_search_limit=25)
max_search_limit = _clamped_int(raw.get("max_search_limit"), 25, 1, 50)
defer_raw = raw.get("defer")
return cls(
@@ -115,9 +77,7 @@ class ToolSearchConfig:
# A list replaces the curated default wholesale; anything else = curated.
defer_tools=(
frozenset(str(n).strip() for n in defer_raw if str(n).strip())
if isinstance(defer_raw, (list, tuple, set)) else None
),
)
if isinstance(defer_raw, (list, tuple, set)) else None))
_TRI_STATE_ALIASES = {"true": "on", "1": "on", "yes": "on", "false": "off", "0": "off", "no": "off"}
@@ -130,16 +90,13 @@ def _tri_state(value: Any) -> str:
return text if text in ("auto", "on", "off") else "auto"
def _safe_int(value: Any, fallback: int) -> int:
try:
return int(value)
except (TypeError, ValueError):
return fallback
def _clamped_int(value: Any, fallback: int, lo: int, hi: int) -> int:
"""``_safe_int`` clamped to ``[lo, hi]`` (the fallback is clamped too)."""
return max(lo, min(hi, _safe_int(value, fallback)))
"""``int(value)`` (or ``fallback`` when unparseable) clamped to ``[lo, hi]``."""
try:
value = int(value)
except (TypeError, ValueError):
value = fallback
return max(lo, min(hi, value))
def _safe_float(value: Any, fallback: float) -> float:
@@ -150,6 +107,7 @@ def _safe_float(value: Any, fallback: float) -> float:
def _config_from_loader(loader_name: str) -> ToolSearchConfig:
"""Tool-search config via ``hermes_cli.config.<loader_name>`` (defaults on any failure)."""
try:
import hermes_cli.config as _cfg_mod
cfg = getattr(_cfg_mod, loader_name)() or {}
@@ -161,20 +119,14 @@ def _config_from_loader(loader_name: str) -> ToolSearchConfig:
def load_config() -> ToolSearchConfig:
"""Load tool-search config from the user config file."""
return _config_from_loader("load_config")
def load_config_readonly() -> ToolSearchConfig:
"""Load tool-search config without copying the cached full config."""
"""Same as ``load_config`` without copying the cached full config."""
return _config_from_loader("load_config_readonly")
# ---------------------------------------------------------------------------
# Tool classification
# ---------------------------------------------------------------------------
def _core_tool_names() -> frozenset[str]:
"""Tool names that never defer by default (lazy import: ``toolsets`` imports
``tools.registry``, so a module-level import would be a cycle)."""
@@ -190,9 +142,8 @@ def _core_tool_names() -> frozenset[str]:
_DIRECT_SURFACE_TOOLSETS = frozenset({"desktop_ui", "project"})
# Curated event-triggered core tools deferred BY DEFAULT — reached for when
# something specific happens, not every-turn working-set tools, so a catalog
# stub suffices. ``tools.tool_search.defer`` replaces this list wholesale
# ([] = legacy everything-eager). Names are POST-rename.
# something specific happens, so a catalog stub suffices. ``tools.tool_search.defer``
# replaces this list wholesale ([] = legacy everything-eager). Names are POST-rename.
# ``clarify`` is deliberately NOT here: A/B showed deferring it collapsed
# structured-clarify usage (18/18 -> 7/18) — the ask-the-user affordance must
# be ambient to fire; a stub is not enough.
@@ -225,25 +176,6 @@ def is_deferrable_tool_name(name: str, defer_tools: Optional[frozenset] = None)
return False
def _describe_classification(
name: str,
defer_tools: Optional[frozenset] = None,
) -> Literal["available", "not_found", "not_deferrable"]:
"""Classify a describe name without treating unknown names as errors."""
entry = _registry_entry(name)
if entry is None:
return "not_found"
if defer_tools is not None and name in defer_tools:
return "available"
if (
name in BRIDGE_TOOL_NAMES
or name in _core_tool_names()
or entry.toolset in _DIRECT_SURFACE_TOOLSETS
):
return "not_deferrable"
return "available"
def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
"""Function names of a tool-defs list (``""`` for a nameless def)."""
return (_fn(td).get("name", "") for td in tool_defs)
@@ -251,8 +183,7 @@ def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
def classify_tools(
tool_defs: List[Dict[str, Any]],
defer_tools: Optional[frozenset] = None,
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
defer_tools: Optional[frozenset] = None) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""Split a tool-defs list into (visible, deferrable). Bridge tools are
dropped (they are re-added after classification)."""
visible: List[Dict[str, Any]] = []
@@ -270,11 +201,6 @@ def _deferrable_in(tool_defs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
return classify_tools(tool_defs, load_config_readonly().effective_defer_tools)[1]
# ---------------------------------------------------------------------------
# Token estimation and threshold gate
# ---------------------------------------------------------------------------
def estimate_tokens_from_schemas(tool_defs: Iterable[Dict[str, Any]]) -> int:
"""Token cost of a tool-defs list via the chars/4 rule (order-of-magnitude
precision is all the activation gate needs)."""
@@ -290,20 +216,14 @@ def estimate_tokens_from_schemas(tool_defs: Iterable[Dict[str, Any]]) -> int:
def should_activate(
config: ToolSearchConfig,
deferrable_tokens: int,
context_length: Optional[int],
) -> bool:
"""``"off"`` never activates; ``"on"``/``"auto"`` activate whenever any
deferrable tool exists. ``"auto"`` is an alias of ``"on"`` reserved for a
future budget-gated mode — do not distinguish them without that design.
``context_length`` is kept in the signature for caller compatibility; the
threshold governs the listing budget, not activation."""
context_length: Optional[int]) -> bool:
"""``"off"`` never activates; ``"on"``/``"auto"`` activate whenever any deferrable
tool exists ("auto" is reserved for a future budget-gated mode — do not distinguish
them without that design). ``context_length`` is kept for caller compatibility."""
return config.enabled != "off" and deferrable_tokens > 0
def listing_token_budget(
config: ToolSearchConfig,
context_length: Optional[int],
) -> int:
def listing_token_budget(config: ToolSearchConfig, context_length: Optional[int]) -> int:
"""``min(listing_max_tokens, threshold_pct% of context)``; unknown context
uses a 10K percentage leg (5% of a typical 200K window)."""
if context_length and context_length > 0:
@@ -313,11 +233,6 @@ def listing_token_budget(
return max(0, min(config.listing_max_tokens, pct_leg))
# ---------------------------------------------------------------------------
# Catalog + BM25 retrieval
# ---------------------------------------------------------------------------
def _bridge_schema(name: str, description: str, properties: Dict[str, Any],
required: List[str]) -> Dict[str, Any]:
"""One OpenAI-style function schema (key order is part of the frozen bytes)."""
@@ -342,8 +257,7 @@ def _search_description(deferred_count: int, listing: Optional[str], listing_for
"tool's description. Follow with "
f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, "
f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this "
"system prompt are already available and do not need to be searched."
)
"system prompt are already available and do not need to be searched.")
if not listing:
return desc
if listing_form == "groups":
@@ -352,33 +266,29 @@ def _search_description(deferred_count: int, listing: Optional[str], listing_for
"through this bridge. For any request in these domains, search "
"here FIRST — do not claim the capability is unavailable and do "
"not substitute a generic tool (terminal/browser) without "
"searching.\n\n" + listing
)
"searching.\n\n" + listing)
desc += (
"\n\nEvery deferred capability is listed below. If a tool name "
"appears here, do NOT claim it is unavailable — load it with "
f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you "
"already see the exact name)."
)
"already see the exact name).")
if listing_form == "mixed":
desc += (
" For servers marked 'names not listed', the tools exist "
f"too — find them with `{TOOL_SEARCH_NAME}` before "
"concluding anything is missing."
)
"concluding anything is missing.")
return desc + "\n\n" + listing
def bridge_tool_schemas(
deferred_count: int,
listing: Optional[str] = None,
listing_form: str = "",
) -> List[Dict[str, Any]]:
listing_form: str = "") -> List[Dict[str, Any]]:
"""Bridge tool schemas injected in place of deferred tools. Kept short —
every byte is paid on every turn. ``listing`` is embedded in the
tool_search description; ``listing_form`` picks the framing (per-tool forms
say "skip search when you see the exact name", the "groups" summary says
which domains exist and that search is mandatory)."""
every byte is paid on every turn. ``listing`` is embedded in the tool_search
description; ``listing_form`` picks the framing (per-tool forms say "skip
search when you see the exact name", "groups" says which domains exist and
that search is mandatory)."""
return [
_bridge_schema(
TOOL_SEARCH_NAME,
@@ -427,14 +337,9 @@ def bridge_tool_schemas(
]
# ---------------------------------------------------------------------------
# Public entry point: assemble tool-defs with optional tool search
# ---------------------------------------------------------------------------
@dataclass
class AssemblyResult:
"""Outcome of one assembly. Useful for tests and observability."""
"""Outcome of one assembly (tests and observability)."""
tool_defs: List[Dict[str, Any]]
activated: bool
@@ -442,8 +347,7 @@ class AssemblyResult:
deferred_tokens: int = 0
threshold_tokens: int = 0
# 0 = passthrough; 1 = bridge + per-tool listing (full/names/mixed);
# 2 = bare bridge / server-summary only — domains stay visible but
# individual tools are reachable only via tool_search.
# 2 = bare bridge / server-summary only (tools reachable only via tool_search).
tier: int = 0
listing_form: str = "none" # "full" | "names" | "mixed" | "groups" | "none"
@@ -452,17 +356,13 @@ def assemble_tool_defs(
tool_defs: List[Dict[str, Any]],
*,
context_length: Optional[int] = None,
config: Optional[ToolSearchConfig] = None,
) -> AssemblyResult:
"""Return the tool-defs list the model should actually see: passthrough
when inactive, otherwise deferrable tools replaced by the three bridge
tools. Idempotent — bridge tools already present are stripped first."""
config: Optional[ToolSearchConfig] = None) -> AssemblyResult:
"""Tool-defs the model should see: passthrough when inactive, else deferrable tools
replaced by the three bridge tools. Idempotent — existing bridge tools are stripped first."""
if config is None:
config = load_config()
incoming = [td for td, name in zip(tool_defs, _tool_def_names(tool_defs))
if name not in BRIDGE_TOOL_NAMES]
visible, deferrable = classify_tools(incoming, config.effective_defer_tools)
if not deferrable:
return AssemblyResult(tool_defs=incoming, activated=False)
@@ -470,44 +370,26 @@ def assemble_tool_defs(
deferrable_tokens = estimate_tokens_from_schemas(deferrable)
if not should_activate(config, deferrable_tokens, context_length):
return AssemblyResult(
tool_defs=incoming,
activated=False,
deferred_count=len(deferrable),
tool_defs=incoming, activated=False, deferred_count=len(deferrable),
deferred_tokens=deferrable_tokens,
threshold_tokens=int((context_length or 0) * (config.threshold_pct / 100.0)),
tier=0,
)
tier=0)
listing, listing_form = None, "none"
listing_budget = listing_token_budget(config, context_length)
if config.listing != "off":
listing, listing_form = build_catalog_listing_with_form(
deferrable, max_tokens=listing_budget)
bridge = bridge_tool_schemas(len(deferrable), listing=listing,
listing_form=listing_form)
bridge = bridge_tool_schemas(len(deferrable), listing=listing, listing_form=listing_form)
tier = 1 if listing_form in ("full", "names", "mixed") else 2
logger.info(
"tool_search activated (tier %d): %d core/visible tools kept, %d deferred "
"(~%d tokens), listing %s (budget ~%d tokens)",
tier, len(visible), len(deferrable), deferrable_tokens,
listing_form, listing_budget,
)
tier, len(visible), len(deferrable), deferrable_tokens, listing_form, listing_budget)
return AssemblyResult(
tool_defs=visible + bridge,
activated=True,
deferred_count=len(deferrable),
deferred_tokens=deferrable_tokens,
threshold_tokens=listing_budget,
tier=tier,
listing_form=listing_form,
)
# ---------------------------------------------------------------------------
# Bridge tool dispatch
# ---------------------------------------------------------------------------
tool_defs=visible + bridge, activated=True, deferred_count=len(deferrable),
deferred_tokens=deferrable_tokens, threshold_tokens=listing_budget,
tier=tier, listing_form=listing_form)
def is_bridge_tool(name: str) -> bool:
@@ -515,9 +397,8 @@ def is_bridge_tool(name: str) -> bool:
def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
"""One record for the response's shared ``tools`` map (held once per tool;
per-query groups carry names only). ``required`` lets the model attempt a
trivial call without a ``tool_describe`` round-trip."""
"""One record for the shared ``tools`` map (per-query groups carry names only);
``required`` lets the model attempt a trivial call without a ``tool_describe`` round-trip."""
schema = entry.schema if isinstance(entry.schema, dict) else {}
fn = schema.get("function")
params = fn.get("parameters") if isinstance(fn, dict) else None
@@ -528,14 +409,12 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
"source": entry.source,
"source_name": entry.source_name,
"description": (entry.description or "")[:400], # cap chatty MCP descriptions
"required": [r[:64] for r in required if isinstance(r, str)][:32],
}
"required": [r[:64] for r in required if isinstance(r, str)][:32]}
def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any]]:
"""Deterministic ``[{name, tool_count}]`` of connected sources, attached to
empty query groups so a lexical miss is not read as a missing capability
(adds nothing to the fixed per-turn prompt)."""
empty query groups so a lexical miss is not read as a missing capability."""
counts = Counter(_listing_group_label(entry.source_name) for entry in catalog)
return [{"name": name, "tool_count": counts[name]} for name in sorted(counts)]
@@ -543,10 +422,8 @@ def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any
def _string_list_arg(
args: Dict[str, Any], key: str, *, dedupe: bool, max_items: int, retry_hint: str,
) -> Tuple[Optional[List[str]], Optional[str]]:
"""Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare
string (a common model slip) counts as a one-item list. Rejects non-list
input, an empty list (after stripping blanks), and more than ``max_items``
(bounds the work one bridge call can request)."""
"""Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare string (a
common model slip) is a one-item list; rejects non-lists, all-blank lists, > ``max_items``."""
raw = args.get(key)
if isinstance(raw, str):
raw = [raw]
@@ -581,21 +458,16 @@ def dispatch_tool_search(args: Dict[str, Any],
"""
if config is None:
config = load_config()
queries, err = _string_list_arg(
args, "queries", dedupe=False, max_items=_MAX_QUERIES_PER_CALL,
retry_hint="Retry with fewer, more targeted queries.")
if err:
return err
raw_limit = args.get("limit")
if raw_limit is None:
limit = config.search_default_limit
else:
limit = _clamped_int(raw_limit, config.search_default_limit, 1, config.max_search_limit)
limit = (config.search_default_limit if raw_limit is None
else _clamped_int(raw_limit, config.search_default_limit, 1, config.max_search_limit))
catalog = build_catalog(_deferrable_in(current_tool_defs))
results: List[Dict[str, Any]] = []
tools_map: Dict[str, Dict[str, Any]] = {}
corpus_stats = _corpus_stats(catalog)
@@ -611,10 +483,8 @@ def dispatch_tool_search(args: Dict[str, Any],
"This query returned no lexical matches, but the sources above "
"are connected and their tools remain available. Retry "
"tool_search with the service name plus a concrete action or "
"object before concluding the capability is unavailable."
)
"object before concluding the capability is unavailable.")
results.append(group)
return json.dumps({
"queries": queries,
"total_available": len(catalog),
@@ -637,7 +507,6 @@ def dispatch_tool_describe(args: Dict[str, Any],
"""
if config is None:
config = load_config_readonly()
names, err = _string_list_arg(
args, "names", dedupe=True, max_items=_MAX_DESCRIBE_NAMES_PER_CALL,
retry_hint="Retry with fewer names per call.")
@@ -646,7 +515,6 @@ def dispatch_tool_describe(args: Dict[str, Any],
deferrable = _deferrable_in(current_tool_defs)
by_name = {name: _fn(td) for td, name in zip(deferrable, _tool_def_names(deferrable)) if name}
tools: Dict[str, Dict[str, Any]] = {}
not_found: List[str] = []
errors: Dict[str, str] = {}
@@ -655,15 +523,13 @@ def dispatch_tool_describe(args: Dict[str, Any],
if fn is not None:
tools[name] = {
"description": fn.get("description", ""),
"parameters": fn.get("parameters", {}),
}
elif _describe_classification(
name, load_config_readonly().effective_defer_tools
) == "not_deferrable":
"parameters": fn.get("parameters", {})}
elif _registry_entry(name) is not None and not is_deferrable_tool_name(
name, load_config_readonly().effective_defer_tools):
# Registered but bridge/core/GUI-surface: a real name, wrong door.
errors[name] = (
f"'{name}' is not a deferrable tool. If you see it in the tools list "
"already, call it directly; otherwise check the spelling against tool_search."
)
"already, call it directly; otherwise check the spelling against tool_search.")
else:
not_found.append(name)
@@ -677,21 +543,18 @@ def dispatch_tool_describe(args: Dict[str, Any],
def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
"""Deferrable tool names in the *pre-assembly* ``tool_defs`` of the session's
toolset scope — the universe ``tool_call`` may reach. Gates both bridge
dispatch and the executor unwrap so a restricted session cannot invoke an
out-of-scope tool via the bridge."""
"""Deferrable names in the *pre-assembly* ``tool_defs`` of the session scope — the
universe ``tool_call`` may reach. Gates bridge dispatch AND the executor unwrap so a
restricted session cannot invoke an out-of-scope tool via the bridge."""
defer_tools = load_config_readonly().effective_defer_tools
return frozenset(
name for name in _tool_def_names(tool_defs)
if name and is_deferrable_tool_name(name, defer_tools)
)
if name and is_deferrable_tool_name(name, defer_tools))
def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any], Optional[str]]:
"""Parse a ``tool_call`` invocation into (underlying_name, args, error_msg);
``(None, {}, msg)`` on parse error. Shared by dispatch, display, and the
trajectory recorder so all three agree on the underlying tool."""
"""Parse a ``tool_call`` invocation -> (underlying_name, args, error_msg); ``(None, {}, msg)``
on error. Shared by dispatch, display and the trajectory recorder so all three agree."""
name = str(args.get("name") or "").strip()
if not name:
return None, {}, "tool_call requires a 'name' argument"
@@ -710,35 +573,16 @@ def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[s
if not is_deferrable_tool_name(name, load_config_readonly().effective_defer_tools):
return None, {}, (
f"'{name}' is not a deferrable tool. If it appears in the model-facing tools "
"list already, call it directly instead of via tool_call."
)
"list already, call it directly instead of via tool_call.")
return name, raw_args, None
__all__ = [
"TOOL_SEARCH_NAME",
"TOOL_DESCRIBE_NAME",
"TOOL_CALL_NAME",
"BRIDGE_TOOL_NAMES",
"ToolSearchConfig",
"CatalogEntry",
"AssemblyResult",
"load_config",
"is_deferrable_tool_name",
"classify_tools",
"estimate_tokens_from_schemas",
"should_activate",
"build_catalog",
"build_catalog_listing",
"build_catalog_listing_with_form",
"listing_token_budget",
"search_catalog",
"bridge_tool_schemas",
"assemble_tool_defs",
"is_bridge_tool",
"dispatch_tool_search",
"dispatch_tool_describe",
"resolve_underlying_call",
"scoped_deferrable_names",
"validate_deferred_call_args",
"TOOL_SEARCH_NAME", "TOOL_DESCRIBE_NAME", "TOOL_CALL_NAME", "BRIDGE_TOOL_NAMES",
"ToolSearchConfig", "CatalogEntry", "AssemblyResult", "load_config",
"is_deferrable_tool_name", "classify_tools", "estimate_tokens_from_schemas",
"should_activate", "build_catalog", "build_catalog_listing_with_form", "listing_token_budget",
"search_catalog", "bridge_tool_schemas", "assemble_tool_defs", "is_bridge_tool",
"dispatch_tool_search", "dispatch_tool_describe", "resolve_underlying_call",
"scoped_deferrable_names", "validate_deferred_call_args",
]
tools/tool_search_catalog.py
+15 -51
View File
@@ -42,8 +42,7 @@ _thread_local = threading.local()
def _stemmer() -> Any:
st = getattr(_thread_local, "stemmer", None)
if st is None:
st = snowballstemmer.stemmer("english")
_thread_local.stemmer = st
st = _thread_local.stemmer = snowballstemmer.stemmer("english")
return st
@@ -54,10 +53,8 @@ def _stem(token: str) -> str:
def _tokenize(text: str) -> List[str]:
"""Lowercase alphanumeric tokens, Snowball-stemmed (English).
Shared by the index path and the query path so "issues" matches ``create_issue``.
"""
"""Lowercase alphanumeric tokens, Snowball-stemmed (English). Shared by the
index and query paths so "issues" matches ``create_issue``."""
if not text:
return []
return [_stem(token.lower()) for token in _TOKEN_RE.findall(text)]
@@ -84,8 +81,7 @@ def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str:
top-level parameter names. Schema bodies are excluded (noise, no recall
gain). The ``mcp__`` prefix is dropped — it is in every MCP document, so
its IDF is ~0. The source label lets a service-name query ("linear") reach
a tool whose own name omits the vendor.
"""
a tool whose own name omits the vendor."""
fn = _fn(td)
name = fn.get("name", "")
if name.startswith("mcp__"):
@@ -126,8 +122,7 @@ def build_catalog(tool_defs: List[Dict[str, Any]]) -> List[CatalogEntry]:
schema=td,
source=source,
source_name=source_name,
_tokens=_tokenize(_entry_search_text(td, source_label)),
))
_tokens=_tokenize(_entry_search_text(td, source_label))))
return catalog
@@ -135,8 +130,8 @@ def _bm25_score(query_tokens: List[str], doc_tokens: List[str],
doc_lengths: List[int], avg_dl: float,
doc_freq: Dict[str, int], n_docs: int,
k1: float = 1.5, b: float = 0.75) -> float:
"""Standard BM25 score for one query against one document (inlined; the
catalog is bounded — typically < 500 tools — so a dependency is not worth it)."""
"""Standard BM25 for one query against one document (inlined; the catalog
is bounded — typically < 500 tools — so a dependency is not worth it)."""
if not doc_tokens:
return 0.0
score = 0.0
@@ -144,12 +139,10 @@ def _bm25_score(query_tokens: List[str], doc_tokens: List[str],
doc_tf = Counter(doc_tokens)
for q in query_tokens:
df = doc_freq.get(q, 0)
if df == 0:
tf = doc_tf.get(q, 0)
if df == 0 or tf == 0:
continue
idf = math.log(1 + (n_docs - df + 0.5) / (df + 0.5))
tf = doc_tf.get(q, 0)
if tf == 0:
continue
norm = tf * (k1 + 1) / (tf + k1 * (1 - b + b * dl / max(avg_dl, 1.0)))
score += idf * norm
return score
@@ -173,22 +166,18 @@ def search_catalog(
query: str,
limit: int = 5,
*,
corpus_stats: Optional[_CorpusStats] = None,
) -> List[CatalogEntry]:
corpus_stats: Optional[_CorpusStats] = None) -> List[CatalogEntry]:
"""Top-``limit`` catalog entries for ``query`` by BM25 (exact name match
ranks first). Falls back to a name-substring match only when NO query
token appears in any document (e.g. "hub" vs ``github_*``); the IDF
variant is strictly positive, so a hit anywhere suppresses the fallback.
"""
variant is strictly positive, so a hit anywhere suppresses the fallback."""
if not catalog or limit <= 0:
return []
query_tokens = _tokenize(query)
if not query_tokens:
return []
if corpus_stats is None:
corpus_stats = _corpus_stats(catalog)
doc_lengths, avg_dl, doc_freq, n_docs = corpus_stats
scored: List[Tuple[float, CatalogEntry]] = []
exact_name = query.strip().lower()
@@ -196,17 +185,12 @@ def search_catalog(
if entry.name.lower() == exact_name:
scored.append((float("inf"), entry))
continue
s = _bm25_score(query_tokens, entry._tokens, doc_lengths, avg_dl,
doc_freq, n_docs)
s = _bm25_score(query_tokens, entry._tokens, *corpus_stats)
if s > 0:
scored.append((s, entry))
if not scored:
ql = query.lower()
for entry in catalog:
if ql in entry.name.lower():
scored.append((0.1, entry))
scored = [(0.1, entry) for entry in catalog if ql in entry.name.lower()]
scored.sort(key=lambda x: x[0], reverse=True)
return [e for _, e in scored[:limit]]
@@ -239,21 +223,8 @@ def _listing_group_label(source_name: str) -> str:
return label[4:] if label.startswith("mcp-") else label
def build_catalog_listing(
deferrable: List[Dict[str, Any]],
*,
max_tokens: int = 4000,
) -> Optional[str]:
"""Render the deferred-catalog manifest; text only (see
:func:`build_catalog_listing_with_form` for the degradation ladder)."""
return build_catalog_listing_with_form(deferrable, max_tokens=max_tokens)[0]
def build_catalog_listing_with_form(
deferrable: List[Dict[str, Any]],
*,
max_tokens: int = 4000,
) -> Tuple[Optional[str], str]:
deferrable: List[Dict[str, Any]], *, max_tokens: int = 4000) -> Tuple[Optional[str], str]:
"""Render the skills-style deferred-catalog manifest: ``- name: short desc``
lines grouped under a heading per source (MCP server / plugin toolset).
@@ -267,9 +238,6 @@ def build_catalog_listing_with_form(
Degradation is PER SERVER (largest first): one huge server must not cost
a small co-attached server its listing.
"""
if not deferrable:
return None, "none"
groups: Dict[str, List[Tuple[str, str]]] = {}
for td in deferrable:
fn = _fn(td)
@@ -280,7 +248,6 @@ def build_catalog_listing_with_form(
# ``_listing_group_label("")`` is "other", so one call covers both.
label = _listing_group_label(_classify_source(name)[1])
groups.setdefault(label, []).append((name, _short_desc(fn.get("description", ""))))
if not groups:
return None, "none"
@@ -310,10 +277,8 @@ def build_catalog_listing_with_form(
text = assemble_if_fits(modes)
if text is not None:
return text, mode
# 3. Per-server degradation: collapse the LARGEST rendered groups first
# (deterministic: size then label) so one oversized server does not
# cost a small co-attached server its per-tool names.
# (deterministic: size then label).
by_size = sorted(groups, key=lambda lbl: (-len(render_group(lbl, "names")), lbl))
for lbl in by_size:
modes[lbl] = "summary"
@@ -321,5 +286,4 @@ def build_catalog_listing_with_form(
if text is not None:
form = "groups" if all(m == "summary" for m in modes.values()) else "mixed"
return text, form
return None, "none"
tools/tool_search_validation.py
+14 -39
View File
@@ -22,20 +22,13 @@ def _schema_for_local_validation(node: Any) -> Any:
return [_schema_for_local_validation(item) for item in node]
if not isinstance(node, dict):
return node
normalized = {}
for key, value in node.items():
if key == "nullable":
continue
# Literal keywords hold instance data, not schemas: copy byte-for-byte.
normalized[key] = (
copy.deepcopy(value)
if key in _SCHEMA_LITERAL_KEYS
else _schema_for_local_validation(value)
)
# Literal keywords hold instance data, not schemas: copy byte-for-byte.
normalized = {
key: (copy.deepcopy(value) if key in _SCHEMA_LITERAL_KEYS
else _schema_for_local_validation(value))
for key, value in node.items() if key != "nullable"}
if node.get("nullable") is not True:
return normalized
schema_type = normalized.get("type")
if isinstance(schema_type, str):
if schema_type != "null":
@@ -45,7 +38,6 @@ def _schema_for_local_validation(node: Any) -> Any:
if "null" not in schema_type:
normalized["type"] = [*schema_type, "null"]
return normalized
# No ``type`` to extend ($ref/combinator): wrap so local refs still resolve
# from the root while null stays an explicit alternative.
return {"anyOf": [normalized, {"type": "null"}]}
@@ -64,8 +56,7 @@ def _schema_has_external_ref(node: Any) -> bool:
return any(
_schema_has_external_ref(value)
for key, value in node.items()
if key not in _SCHEMA_LITERAL_KEYS
)
if key not in _SCHEMA_LITERAL_KEYS)
def _validation_path(error: Any) -> str:
@@ -83,12 +74,8 @@ def _validation_path(error: Any) -> str:
def _validation_error(message: str, *, path: str, constraint: str, parameters: Any) -> str:
return tool_error(
message,
path=path,
constraint=constraint,
parameters=parameters,
hint="Retry tool_call with 'arguments' matching the parameters schema above.",
)
message, path=path, constraint=constraint, parameters=parameters,
hint="Retry tool_call with 'arguments' matching the parameters schema above.")
def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str]:
@@ -117,23 +104,18 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
return _validation_error(
f"tool_call to '{name}' is missing required argument(s): "
f"{', '.join(missing)}. The tool was NOT invoked.",
path="arguments", constraint="required", parameters=params,
)
path="arguments", constraint="required", parameters=params)
validation_schema = _schema_for_local_validation(params)
if _schema_has_external_ref(validation_schema):
logger.debug(
"Skipping local deferred-argument validation for %s: external $ref",
name,
)
logger.debug("Skipping local deferred-argument validation for %s: external $ref", name)
return None
# Validate the repaired shape dispatch will see; copy because
# coerce_tool_args may normalize in place (dispatch re-coerces canonically).
candidate_args = dict(args)
try:
from model_tools import coerce_tool_args
candidate_args = coerce_tool_args(name, candidate_args)
candidate_args = coerce_tool_args(name, dict(args))
except Exception:
logger.debug("Deferred-argument coercion failed for %s", name, exc_info=True)
candidate_args = dict(args)
@@ -142,20 +124,14 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
from jsonschema.exceptions import best_match
from jsonschema.validators import validator_for
except ImportError:
logger.debug(
"jsonschema unavailable; keeping required-only validation for %s",
name,
)
logger.debug("jsonschema unavailable; keeping required-only validation for %s", name)
return None
validator_cls = validator_for(validation_schema)
validator_cls.check_schema(validation_schema)
validation_error = best_match(
validator_cls(validation_schema).iter_errors(candidate_args)
)
validation_error = best_match(validator_cls(validation_schema).iter_errors(candidate_args))
if validation_error is None:
return None
path = _validation_path(validation_error)
constraint = str(getattr(validation_error, "validator", None) or "schema")
detail = re.sub(r"\s+", " ", str(validation_error.message)).strip()
@@ -164,8 +140,7 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
return _validation_error(
f"tool_call to '{name}' failed argument validation at {path} "
f"({constraint}): {detail}. The tool was NOT invoked.",
path=path, constraint=constraint, parameters=params,
)
path=path, constraint=constraint, parameters=params)
except Exception: # pragma: no cover — never block dispatch on validator bugs
logger.debug("validate_deferred_call_args failed for %s", name, exc_info=True)
return None
tools/tour_tool.py
+8 -20
View File
@@ -1,17 +1,11 @@
#!/usr/bin/env python3
"""Guided tour (highlight + narrate UI elements) in the Hermes desktop GUI.
Generic, no baked-in tours: the agent discovers targets (``action="targets"``)
then highlights by CSS selector one step at a time (``show``) or hands over a
step list the user pages with Next/Prev (``start``). Surfaces ``app`` (Hermes'
own DOM) and ``preview`` (the in-app browser page) share the renderer's driver.js.
Round-trips through the gateway blocking-prompt bridge (``tour.request`` /
``tour.respond``) so the agent learns whether the selector matched.
Lives in ``desktop_ui`` (GUI sessions only) and withdraws itself when the user
turns tours off: a tour takes the whole screen, so "off" must mean the model is
never told the tool exists rather than being offered a call that fails.
"""
"""Guided tour (highlight + narrate UI elements) in the Hermes desktop GUI. Generic:
the agent discovers targets (``action="targets"``), then highlights one step at a
time (``show``) or hands over a step list the user pages (``start``). Round-trips
through the gateway blocking-prompt bridge (``tour.request``/``tour.respond``) so the
agent learns whether the selector matched. Lives in ``desktop_ui`` and withdraws
itself when tours are off: a tour takes the whole screen, so "off" must mean the
model is never told the tool exists rather than offered a call that fails."""
import json
from typing import Callable, Optional
@@ -170,13 +164,7 @@ registry.register(
schema=TOUR_SCHEMA,
handler=lambda args, **kw: tour_tool(
action=args.get("action", ""),
surface=args.get("surface"),
selector=args.get("selector"),
title=args.get("title"),
text=args.get("text"),
side=args.get("side"),
steps=args.get("steps"),
step_index=args.get("step_index"),
**{k: args.get(k) for k in ("surface", "selector", "title", "text", "side", "steps", "step_index")},
callback=kw.get("callback"),
),
check_fn=check_tours_enabled,