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

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

Browse Source
This commit is contained in:
Teknium
2026-09-03 01:21:16 -07:00
parent 0f66267bf4 e0df9656bb
commit b33d84bb21
11 changed files with 483 additions and 832 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/spill_safety.py
+21 -35
View File
@@ -1,11 +1,10 @@
"""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."""
"""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``."""
from __future__ import annotations
@@ -16,15 +15,14 @@ from typing import IO
__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.
# O_NOFOLLOW is POSIX-only; on Windows O_EXCL alone already refuses every pre-existing path.
_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.
Raises ``OSError`` if the leaf is not a real directory."""
"""Create ``path`` (and parents) as a directory, refusing symlinks. ``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)
@@ -35,17 +33,12 @@ def ensure_spill_dir(path: Path, *, private: bool = True) -> Path:
return path
def open_exclusive(
path: Path,
*,
private: bool = True,
overwrite: bool = False,
encoding: str = "utf-8",
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."""
def open_exclusive(path: Path, *, private: bool = True, overwrite: bool = False,
encoding: str = "utf-8", 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."""
path = Path(path)
if overwrite:
try:
@@ -65,16 +58,9 @@ def open_exclusive(
raise
def write_text_exclusive(
path: Path,
text: str,
*,
private: bool = True,
overwrite: bool = False,
encoding: str = "utf-8",
errors: str = "strict") -> None:
def write_text_exclusive(path: Path, text: str, *, private: bool = True, overwrite: bool = False,
encoding: str = "utf-8", 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
) as fh:
with open_exclusive(path, private=private, overwrite=overwrite, encoding=encoding,
errors=errors) as fh:
fh.write(text)
tools/tip_tool.py
+6 -14
View File
@@ -1,4 +1,3 @@
#!/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) with no scrim/spotlight/paging.
Fire-and-forget: a tip is not a question, so blocking on a round-trip would stall the
@@ -19,15 +18,12 @@ def tip_tool(text: str, selector: str, title: str = "", side: str = "") -> str:
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 "
"what's on screen and prefer a target reporting stable: true."
)
return tool_error("tip needs a selector to point at. Call tour(action='targets') to see "
"what's on screen and prefer a target reporting stable: true.")
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})
payload = {"selector": selector, "text": text,
**{k: v for k, v in (("title", title), ("side", side)) if v}}
try:
ok = desktop_ui.emit("tip.show", payload)
except Exception as exc:
@@ -79,11 +75,7 @@ def check_tips_enabled() -> bool:
registry.register(
name="show_tip",
toolset="desktop_ui",
schema=TIP_SCHEMA,
name="show_tip", toolset="desktop_ui", schema=TIP_SCHEMA, check_fn=check_tips_enabled,
handler=lambda args, **kw: tip_tool(
**{k: args.get(k, "") for k in ("text", "selector", "title", "side")}),
check_fn=check_tips_enabled,
emoji="💡",
)
emoji="💡")
tools/todo_tool.py
+56 -98
View File
@@ -1,39 +1,31 @@
#!/usr/bin/env python3
"""Todo tool: in-memory, revisioned task list for multi-step work.
State lives on the AIAgent (one per session), is re-injected after context
compression, and every write bumps a monotonic revision so UI clients can
reject stale updates. One ``todo_list`` tool: pass ``todos`` to write, omit to
read; every call returns the full list. No system-prompt mutation.
"""
"""Todo tool: in-memory, revisioned task list for multi-step work. State lives on the
AIAgent (one per session), is re-injected after context compression, and every write bumps
a monotonic revision so UI clients can reject stale updates. One ``todo_list`` tool: pass
``todos`` to write, omit to read; every call returns the full list. No system-prompt mutation."""
import json
from typing import Any, Dict, List, Optional
VALID_STATUSES = {"pending", "in_progress", "completed", "cancelled"}
# The list is re-read after every compression (format_for_injection), so
# unbounded content/count would defeat the compression it rides through. Caps
# apply equally to model-authored items and caller-replayed API history.
# The list is re-read after every compression (format_for_injection), so unbounded
# content/count would defeat the compression it rides through. Caps apply equally to
# model-authored items and caller-replayed API history.
MAX_TODO_CONTENT_CHARS = 4000
MAX_TODO_ITEMS = 256
# Max single todo tool-result payload accepted during history hydration, so a
# forged oversized result is dropped before parsing (AIAgent._hydrate_todo_store).
# Max single todo tool-result payload accepted during history hydration, so a forged
# oversized result is dropped before parsing (AIAgent._hydrate_todo_store).
MAX_TODO_RESULT_CHARS = 512_000
_TRUNCATION_MARKER = "… [truncated]"
# Persisted as ordinary message content; ContextCompressor keys on this stable
# header to tell the synthetic post-compaction row from a real user message.
TODO_INJECTION_HEADER = (
"[Your active task list was preserved across context compression]"
)
# Persisted as ordinary message content; ContextCompressor keys on this stable header to
# tell the synthetic post-compaction row from a real user message.
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."""
"""In-memory todo list, one per AIAgent. List position is priority; items are
``{id, content, status, parent?}`` — ``parent`` nests a subtask."""
def __init__(self):
self._items: List[Dict[str, str]] = []
@@ -50,8 +42,7 @@ class TodoStore:
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.
del self._items[MAX_TODO_ITEMS:]
del self._items[MAX_TODO_ITEMS:] # keep the priority head; replays can't grow unbounded
self._sanitize_parents(self._items)
if self._items != before:
self._revision += 1
@@ -63,7 +54,7 @@ class TodoStore:
for t in self._dedupe_by_id(todos):
item_id = str(t.get("id", "")).strip()
if not item_id:
continue # Can't merge without an id
continue # can't merge without an id
cur = existing.get(item_id)
if cur is None:
validated = self._validate(t)
@@ -72,35 +63,26 @@ class TodoStore:
continue
if t.get("content"):
cur["content"] = self._cap_content(str(t["content"]).strip())
if t.get("status"):
status = str(t["status"]).strip().lower()
if status in VALID_STATUSES:
cur["status"] = status
if t.get("status") and str(t["status"]).strip().lower() in VALID_STATUSES:
cur["status"] = str(t["status"]).strip().lower()
if "parent" in t:
parent = str(t["parent"] or "").strip()
if parent:
cur["parent"] = parent
else:
cur.pop("parent", None)
# Rebuild preserving original order for existing items.
seen = set()
rebuilt = []
for item in self._items:
current = existing.get(item["id"], item)
if current["id"] not in seen:
rebuilt.append(current)
seen.add(current["id"])
self._items = self._normalize_order(rebuilt)
# Rebuild preserving original order for existing items (first occurrence wins).
rebuilt = {item["id"]: existing.get(item["id"], item) for item in self._items}
self._items = self._normalize_order(list(rebuilt.values()))
def read(self) -> List[Dict[str, str]]:
"""Return a copy of the current list."""
return [item.copy() for item in self._items]
def has_items(self) -> bool:
return bool(self._items)
def snapshot(self) -> Dict[str, Any]:
"""Return the full state clients can reconcile atomically."""
"""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]]:
@@ -113,20 +95,16 @@ class TodoStore:
return self.read()
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."""
"""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."""
if not self._items:
return None
children: Dict[str, List[Dict[str, str]]] = {}
roots: List[Dict[str, str]] = []
for item in self._items:
parent = item.get("parent")
if parent:
children.setdefault(parent, []).append(item)
else:
roots.append(item)
if item.get("parent"):
children.setdefault(item["parent"], []).append(item)
def render(item: Dict[str, str], depth: int, out: List[str]) -> bool:
kid_lines: List[str] = []
@@ -136,16 +114,15 @@ class TodoStore:
keep = item["status"] in _ACTIVE_STATUSES or has_active_kid
if keep:
marker = _STATUS_MARKERS.get(item["status"], "[?]")
out.append(
f"{' ' * depth}- {marker} {item['id']}. "
f"{item['content']} ({item['status']})"
)
out.append(f"{' ' * depth}- {marker} {item['id']}. "
f"{item['content']} ({item['status']})")
out.extend(kid_lines)
return keep
lines = [TODO_INJECTION_HEADER]
for item in roots:
render(item, 0, lines)
for item in self._items:
if not item.get("parent"):
render(item, 0, lines)
return "\n".join(lines) if len(lines) > 1 else None
@staticmethod
@@ -157,16 +134,15 @@ class TodoStore:
@staticmethod
def _validate(item: Dict[str, Any]) -> Dict[str, str]:
"""Normalize one item to ``{id, content, status, parent?}`` with placeholders for missing fields."""
"""Normalize one item to ``{id, content, status, parent?}`` (placeholders when missing)."""
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}
result = {"id": item_id,
"content": TodoStore._cap_content(content) if content else "(no description)",
"status": status if status in VALID_STATUSES else "pending"}
parent = str(item.get("parent") or "").strip()
if parent and parent != item_id:
result["parent"] = parent
@@ -180,8 +156,7 @@ class TodoStore:
if item.get("parent") and item["parent"] not in by_id:
item.pop("parent", None)
for item in items:
seen = {item["id"]}
node = item
seen, node = {item["id"]}, item
while node.get("parent"):
if node["parent"] in seen:
item.pop("parent", None)
@@ -193,22 +168,17 @@ class TodoStore:
def _dedupe_by_id(todos: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Collapse duplicate ids, keeping the last occurrence in its position."""
last_index: Dict[str, int] = {}
for i, item in enumerate(todos):
if not isinstance(item, dict):
# Non-dict items get a synthetic key so _validate can handle them
last_index[f"__invalid_{i}"] = i
continue
last_index[str(item.get("id", "")).strip() or "?"] = i
for i, item in enumerate(todos): # non-dicts get a synthetic key; _validate handles them
key = str(item.get("id", "")).strip() if isinstance(item, dict) else f"__invalid_{i}"
last_index[key 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."""
if any(item.get("parent") for item in items):
return items
"""Lift the in_progress step ahead of any earlier pending placeholder. Nested lists
keep authored order — reordering would tear a subtask from its siblings."""
statuses = [item["status"] for item in items]
if "in_progress" not in statuses:
if any(item.get("parent") for item in items) or "in_progress" not in statuses:
return items
active_index = statuses.index("in_progress")
if "pending" not in statuses[:active_index]:
@@ -218,15 +188,14 @@ class TodoStore:
return normalized
def todo_tool(
todos: Optional[List[Dict[str, Any]]] = None,
merge: bool = False,
store: Optional[TodoStore] = None,
) -> str:
"""Write ``todos`` (replace or ``merge`` by id) or read when None; returns list + summary JSON."""
def todo_tool(todos: Optional[List[Dict[str, Any]]] = None, merge: bool = False,
store: Optional[TodoStore] = None) -> str:
"""Write ``todos`` (replace, or ``merge`` by id) or read when None -> list + summary JSON."""
if store is None:
return tool_error("TodoStore not initialized")
if todos is not None:
if todos is None:
items = store.read()
else:
if isinstance(todos, str): # LLMs sometimes send a JSON string instead of a list
try:
todos = json.loads(todos)
@@ -235,17 +204,11 @@ def todo_tool(
if not isinstance(todos, list):
return tool_error(f"todos must be a list, got {type(todos).__name__}")
items = store.write(todos, merge)
else:
items = store.read()
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"],
"summary": summary,
}, ensure_ascii=False)
return json.dumps({"todos": items, "revision": store.snapshot()["revision"],
"summary": summary}, ensure_ascii=False)
def check_todo_requirements() -> bool:
@@ -253,8 +216,8 @@ def check_todo_requirements() -> bool:
return True
# Behavioral guidance is baked into the (static, cached) description; item
# shape and merge semantics live ONLY in the parameter schema.
# Behavioral guidance is baked into the (static, cached) description; item shape and merge
# semantics live ONLY in the parameter schema.
TODO_SCHEMA = {
"name": "todo_list",
"description": (
@@ -311,15 +274,10 @@ TODO_SCHEMA = {
}
# --- Registry ---
from tools.registry import registry, tool_error
registry.register(
name="todo_list",
toolset="todo",
schema=TODO_SCHEMA,
name="todo_list", toolset="todo", schema=TODO_SCHEMA, check_fn=check_todo_requirements,
handler=lambda args, **kw: todo_tool(
todos=args.get("todos"), merge=args.get("merge", False), store=kw.get("store")),
check_fn=check_todo_requirements,
emoji="📋",
)
emoji="📋")
tools/tool_backend_helpers.py
+75 -124
View File
@@ -10,8 +10,6 @@ from typing import Any, Dict, Optional
from utils import is_truthy_value
logger = logging.getLogger(__name__)
_DEFAULT_BROWSER_PROVIDER = "local"
_DEFAULT_MODAL_MODE = "auto"
_VALID_MODAL_MODES = {"auto", "direct", "managed"}
@@ -23,7 +21,6 @@ def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
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
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
@@ -31,24 +28,20 @@ def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool:
return False
def nous_tool_gateway_unavailable_message(
capability: str = "the Nous Tool Gateway", *, force_fresh: bool = False) -> str:
def nous_tool_gateway_unavailable_message(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 (
format_nous_portal_entitlement_message,
get_nous_portal_account_info,
)
account_info = get_nous_portal_account_info(force_fresh=force_fresh)
message = format_nous_portal_entitlement_message(account_info, capability=capability)
format_nous_portal_entitlement_message, get_nous_portal_account_info)
message = format_nous_portal_entitlement_message(
get_nous_portal_account_info(force_fresh=force_fresh), 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.")
return (f"{capability} is unavailable. Run `hermes model` to refresh your "
"Nous Portal login and billing status.")
def normalize_browser_cloud_provider(value: object | None) -> str:
@@ -68,48 +61,38 @@ normalize_modal_mode = coerce_modal_mode
def has_direct_modal_credentials() -> bool:
"""Return True when direct Modal credentials/config are available."""
if os.getenv("MODAL_TOKEN_ID") and os.getenv("MODAL_TOKEN_SECRET"):
return True
try:
modal_file_exists = (Path.home() / ".modal.toml").exists()
except (PermissionError, OSError):
modal_file_exists = False
return bool(
(os.getenv("MODAL_TOKEN_ID") and os.getenv("MODAL_TOKEN_SECRET")) or modal_file_exists)
return (Path.home() / ".modal.toml").exists()
except OSError: # includes PermissionError on Path.home()
return False
def resolve_modal_backend_state(
modal_mode: object | None,
*,
has_direct: bool,
managed_ready: bool,
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."""
def resolve_modal_backend_state(modal_mode: object | None, *, has_direct: bool,
managed_ready: bool,
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_ok = managed_enabled and managed_ready
if requested_mode == "managed":
selected_backend = "managed" if managed_ok else None
elif requested_mode == "direct":
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": requested_mode == "managed" and not managed_enabled,
"selected_backend": selected_backend}
exclusive = {"managed": "managed" if managed_ok else None,
"direct": "direct" if has_direct else None}
selected_backend = exclusive.get(
requested_mode, "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": requested_mode == "managed" and not managed_enabled,
"selected_backend": selected_backend}
def _scoped_credential(name: str) -> str:
"""Read a credential env var under the active profile secret scope. Raw env
fallback only if ``agent.secret_scope`` cannot import — a packaging edge must
never leave the caller without a key."""
"""Read a credential env var under the active profile secret scope; raw env fallback only
if ``agent.secret_scope`` cannot import (a packaging edge must never lose the key)."""
try:
from agent.secret_scope import get_secret
return (get_secret(name, "") or "").strip()
except Exception: # pragma: no cover — secret_scope is in-repo
return (os.getenv(name, "") or "").strip()
@@ -119,53 +102,38 @@ 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:
"""Resolve a voice-provider API key (single owner for STT/TTS lookup).
Order: explicit ``config_value`` -> profile secret scope / env -> ``.env``
via ``env_getter`` (or ``hermes_cli.config.get_env_value``) -> credential
pool for ``provider_id``. Under an active multiplex turn the profile scope
is authoritative: a miss returns ``""`` rather than borrowing another
profile's env or pool. Never raises.
"""
value = str(config_value or "").strip()
if value:
return value
key = _scoped_credential(env_var)
def resolve_provider_secret(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`` via ``env_getter`` (or
``hermes_cli.config.get_env_value``) -> credential pool for ``provider_id``. Under an
active multiplex turn the profile scope is authoritative: a miss returns ``""`` rather
than borrowing another profile's env or pool. Never raises."""
key = str(config_value or "").strip() or _scoped_credential(env_var)
if key:
return key
try:
from agent.secret_scope import is_multiplex_active
if is_multiplex_active():
return ""
except Exception: # pragma: no cover — secret_scope is in-repo
pass
key = (str(env_getter(env_var) or "").strip() if env_getter is not None
else _dotenv_value(env_var))
key = str(env_getter(env_var) or "").strip() if env_getter else _dotenv_value(env_var)
if key or not provider_id:
return key
try:
from agent.credential_pool import load_pool
# config.yaml ``providers.<name>`` entries are pooled under ``custom:<name>``.
for pool_key in (provider_id, f"custom:{provider_id}"):
pool = load_pool(pool_key)
if pool is None or not pool.has_credentials():
continue
entry = pool.peek()
if entry is None:
continue
key = str(
getattr(entry, "runtime_api_key", "") or getattr(entry, "access_token", "") or ""
).strip()
entry = pool.peek() if pool is not None and pool.has_credentials() else None
key = str(getattr(entry, "runtime_api_key", "") or getattr(entry, "access_token", "")
or "").strip()
if key:
return key
except Exception as exc:
@@ -174,12 +142,11 @@ 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."""
return (
resolve_provider_secret("VOICE_TOOLS_OPENAI_KEY", "")
or resolve_provider_secret("OPENAI_API_KEY", "openai-api"))
"""Prefer VOICE_TOOLS_OPENAI_KEY, else OPENAI_API_KEY (scope-aware, pool fallback for the
latter). Must go through the secret scope: 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"))
def prefers_gateway(config_section: str) -> bool:
@@ -187,22 +154,18 @@ def prefers_gateway(config_section: str) -> bool:
try:
from hermes_cli.config import load_config
section = (load_config() or {}).get(config_section)
if isinstance(section, dict):
return is_truthy_value(section.get("use_gateway"), default=False)
return isinstance(section, dict) and is_truthy_value(section.get("use_gateway"))
except Exception:
pass
return False
return False
# Provider value the managed "Nous Subscription" picker rows write for every
# category; any other name = that vendor direct; no key = legacy autodetect.
# Provider value the managed "Nous Subscription" picker rows write for every category;
# any other name = that vendor direct; no key = legacy autodetect.
NOUS_MANAGED_PROVIDER = "nous"
# Per-capability keys that also count as "this category has been configured".
_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.
# 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",)}
_DEFAULT_NAME_KEYS = ("provider", "backend", "cloud_provider")
@@ -211,77 +174,65 @@ def _raw_section(section: str) -> Dict[str, Any] | None:
"""The RAW (unmerged) config.yaml mapping for ``section``, or None."""
try:
from hermes_cli.config import read_raw_config_readonly
cfg = read_raw_config_readonly() or {}
raw = cfg.get(section) if isinstance(cfg, dict) else None
return raw if isinstance(raw, dict) else None
except Exception:
return None
return raw if isinstance(raw, dict) else None
def read_selection(section: str) -> str | None:
"""THE single runtime read of the persisted `hermes tools` selection.
Returns ``"nous"`` (managed gateway row), a vendor name (direct, own
credentials), or ``None`` (never configured -> legacy autodetect allowed).
Reads the RAW config.yaml so key presence means "actually written", not
"schema default"; a raw ``local`` is therefore a real user selection.
Legacy shim: ``use_gateway: true`` was only ever written by the managed
row, so it maps to ``"nous"`` regardless of the name key. Never raises.
"""
"""THE single runtime read of the persisted `hermes tools` selection: ``"nous"`` (managed
gateway row), a vendor name (direct, own credentials), or ``None`` (never configured ->
legacy autodetect allowed). Reads the RAW config.yaml so key presence means "actually
written", not "schema default"; a raw ``local`` is therefore a real user selection.
Legacy shim: ``use_gateway: true`` was only ever written by the managed row, so it maps
to ``"nous"`` regardless of the name key. Never raises."""
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):
if is_truthy_value(raw.get("use_gateway")):
return NOUS_MANAGED_PROVIDER
for key in _SELECTION_NAME_KEYS.get(section, _DEFAULT_NAME_KEYS):
value = raw.get(key)
if value is not None:
text = str(value).strip().lower()
if text:
return text
text = str(raw.get(key)).strip().lower() if raw.get(key) is not None else ""
if text:
return text
# use_gateway: false with no name key is not a usable selection shape;
# per-capability web keys still count as configured via selection_exists().
return None
def selection_exists(section: str) -> bool:
"""True when ANY selection signal was ever written for the section
(wider than read_selection: per-capability web keys count too)."""
"""True when ANY selection signal was ever written for the section (wider than
read_selection: per-capability web keys count too)."""
if read_selection(section) is not None:
return True
extra = _EXTRA_SELECTION_KEYS.get(section, ())
raw = _raw_section(section) if extra else None
if raw is None:
return False
return any(str(raw.get(key) or "").strip() for key in extra)
return raw is not None and any(str(raw.get(key) or "").strip() for key in extra)
# 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". 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 (...)"},
# 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". Used by the startup config check and selection_error(); add removals here, never as
# one-off string checks: "web": {"<name>": "the <Name> backend was removed in vX (...)"}
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()."""
normalized = (name or "").strip().strip("'\"").lower()
return REMOVED_BACKENDS.get(section, {}).get(normalized)
"""Explanation for a backend that used to ship in-tree, or None. ``name`` tolerates the
quoted form callers pass to selection_error()."""
return REMOVED_BACKENDS.get(section, {}).get((name or "").strip().strip("'\"").lower())
def selection_error(section: str, selection_name: str, failure: str) -> str:
"""The uniform honest-error contract for a selected-but-broken provider."""
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.")
return (f"{section} is configured to use {selection_name} (set via hermes "
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."""
"""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."""
return bool(_scoped_credential("FAL_KEY") or _dotenv_value("FAL_KEY"))
tools/tool_output_limits.py
+10 -21
View File
@@ -11,9 +11,7 @@ from typing import Any, Dict
DEFAULT_MAX_BYTES = 50_000 # terminal_tool.MAX_OUTPUT_CHARS
DEFAULT_MAX_LINES = 2000 # file_operations.MAX_LINES
DEFAULT_MAX_LINE_LENGTH = 2000 # file_operations.MAX_LINE_LENGTH
# Process-lifetime cache: avoids re-reading config.yaml on every tool call.
_cached_limits: dict | None = None
_cached_limits: dict | None = None # process-lifetime: no config.yaml re-read per tool call
def _coerce_int(value: Any, default: int, minimum: int) -> int:
@@ -26,13 +24,12 @@ def _coerce_int(value: Any, default: int, minimum: int) -> int:
def _coerce_positive_int(value: Any, default: int) -> int:
"""Return ``value`` as a positive int, or ``default`` on any issue."""
return _coerce_int(value, default, 1)
return _coerce_int(value, default, 1) # positive int, or ``default`` on any issue
def get_tool_output_limits() -> Dict[str, int]:
"""Resolved ``{max_bytes, max_lines, max_line_length}``; never raises. Cached
for the process — ``_reset_tool_output_limits_cache()`` forces a fresh read."""
"""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
@@ -40,16 +37,15 @@ def get_tool_output_limits() -> Dict[str, int]:
from hermes_cli.config import load_config
cfg = load_config() or {}
section = cfg.get("tool_output") if isinstance(cfg, dict) else None
if not isinstance(section, dict):
section = {}
except Exception:
section = None
if not isinstance(section, dict):
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
@@ -59,13 +55,6 @@ def _reset_tool_output_limits_cache() -> None:
_cached_limits = None
def get_max_bytes() -> int:
return get_tool_output_limits()["max_bytes"]
def get_max_lines() -> int:
return get_tool_output_limits()["max_lines"]
def get_max_line_length() -> int:
return get_tool_output_limits()["max_line_length"]
def get_max_bytes() -> int: return get_tool_output_limits()["max_bytes"]
def get_max_lines() -> int: return get_tool_output_limits()["max_lines"]
def get_max_line_length() -> int: return get_tool_output_limits()["max_line_length"]
tools/tool_result_storage.py
+70 -106
View File
@@ -1,11 +1,9 @@
"""Tool result persistence -- preserves large outputs instead of truncating.
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``."""
"""Tool result persistence -- preserves large outputs instead of truncating. 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
@@ -34,13 +32,12 @@ _spillover_pruned_once = False
def get_spillover_dir():
"""Return $HERMES_HOME/cache/spillover as a Path (not created)."""
from hermes_constants import get_hermes_home
return get_hermes_home() / SPILLOVER_SUBDIR
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 the gateway housekeeping loop prunes hourly."""
"""Delete spillover files older than *max_age_hours*; returns count removed (same
contract as the ``cleanup_*_cache`` helpers the gateway housekeeping loop runs hourly)."""
cutoff = time.time() - (max_age_hours * 3600)
removed = 0
try:
@@ -53,7 +50,7 @@ def cleanup_spillover_cache(max_age_hours: int = SPILLOVER_MAX_AGE_HOURS) -> int
f.unlink()
removed += 1
except OSError:
continue
pass
return removed
@@ -65,22 +62,19 @@ def _prune_spillover_once() -> None:
return
_spillover_pruned_once = True
try:
removed = cleanup_spillover_cache()
if removed:
if removed := cleanup_spillover_cache():
logger.debug("Pruned %d expired spillover file(s)", removed)
except Exception as exc:
logger.debug("Spillover prune failed: %s", exc)
def _is_host_side_env(env) -> bool:
"""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."""
"""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."""
if env is None:
return True
try:
from tools.environments.local import LocalEnvironment
return isinstance(env, LocalEnvironment)
except Exception:
return False
@@ -101,26 +95,23 @@ def _write_to_spillover(content: str, filename: str):
def _sandbox_visible_spillover_path(host_path: str, env) -> str | None:
"""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."""
"""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
visible = to_agent_visible_cache_path(host_path)
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:
if (sync_manager := getattr(env, "_sync_manager", None)) is not None:
sync_manager.sync(force=True)
except Exception as exc:
logger.debug("Spillover sync failed: %s", exc)
try:
if env.execute(f"test -r {shlex.quote(visible)}", timeout=15).get("returncode", 1) == 0:
return visible
except Exception as exc:
logger.debug("Spillover readability probe failed: %s", exc)
@@ -130,15 +121,13 @@ 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."""
get_temp_dir = getattr(env, "get_temp_dir", None)
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
return f"{temp_dir.rstrip('/') or '/'}/hermes-results" if temp_dir else STORAGE_DIR
def _safe_result_filename(tool_use_id: str) -> str:
@@ -146,9 +135,7 @@ 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
safe_stem = safe_stem or "tool_result"
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"
@@ -160,26 +147,21 @@ def generate_preview(content: str, max_chars: int = DEFAULT_PREVIEW_SIZE_CHARS)
"""Truncate at last newline within max_chars. Returns (preview, has_more)."""
if len(content) <= max_chars:
return content, False
truncated = content[:max_chars]
last_nl = truncated.rfind("\n")
if last_nl > max_chars // 2:
truncated = truncated[:last_nl + 1]
return truncated, True
last_nl = content.rfind("\n", 0, max_chars)
return content[:last_nl + 1 if last_nl > max_chars // 2 else max_chars], True
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 one argv element at 128 KB, so a heredoc-in-command silently failed
for exactly the oversized results persistence exists to handle."""
"""Write content into the sandbox via env.execute(); True on success. Content goes through
stdin, not the command string: Linux ``MAX_ARG_STRLEN`` caps one argv element at 128 KB,
so a heredoc-in-command silently failed for exactly the oversized results this handles."""
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)
return result.get("returncode", 1) == 0
return env.execute(cmd, timeout=30, stdin_data=content).get("returncode", 1) == 0
def _build_persisted_message(
preview: str, has_more: bool, original_size: int, file_path: str) -> str:
def _build_persisted_message(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"
@@ -200,83 +182,65 @@ _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."""
if not isinstance(content, str) or PERSISTED_OUTPUT_TAG not in content:
return None
match = _PERSISTED_PATH_RE.search(content)
"""File path from a <persisted-output> block, or None (lets the result-reference stubbing
guard in agent/tool_guardrails.py carry the spillover path instead of leaving it dangling)."""
match = (_PERSISTED_PATH_RE.search(content)
if isinstance(content, str) and PERSISTED_OUTPUT_TAG in content else None)
return match.group(1).strip() if match else None
def maybe_persist_tool_result(
content: str,
tool_name: str,
tool_use_id: str,
env=None,
config: BudgetConfig = DEFAULT_BUDGET,
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."""
effective_threshold = threshold if threshold is not None else config.resolve_threshold(tool_name)
if effective_threshold == float("inf") or len(content) <= effective_threshold:
def maybe_persist_tool_result(content: str, tool_name: str, tool_use_id: str, env=None,
config: BudgetConfig = DEFAULT_BUDGET,
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."""
if threshold is None:
threshold = config.resolve_threshold(tool_name)
if threshold == float("inf") or len(content) <= threshold:
return content
filename = _safe_result_filename(tool_use_id)
preview, has_more = generate_preview(content, max_chars=config.preview_size)
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)
logger.info("Persisted large tool result: %s (%s, %d chars -> %s%s)",
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.
host_path = _write_to_spillover(content, filename)
if _is_host_side_env(env):
if host_path is not None:
return _persisted(host_path)
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}]")
host_side = _is_host_side_env(env)
if host_side and host_path is not None:
return _persisted(host_path)
if not host_side:
# Remote backend: reference the mounted/synced path when the sandbox can actually read
# it, else write into the sandbox temp dir (containers without the spillover mount).
visible = _sandbox_visible_spillover_path(host_path, env) if host_path else None
if visible is not None:
return _persisted(visible, f" [host: {host_path}]")
remote_path = f"{_resolve_storage_dir(env)}/{filename}"
try:
if _write_to_sandbox(content, remote_path, env):
return _persisted(remote_path)
except Exception as exc:
logger.warning("Sandbox write failed for %s: %s", tool_use_id, exc)
logger.info(
"Inline-truncating large tool result: %s (%d chars, no sandbox write)",
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.]")
logger.info("Inline-truncating large tool result: %s (%d chars, no sandbox write)",
tool_name, len(content))
return (f"{preview}\n\n[Truncated: tool response was {len(content):,} chars. "
"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: 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):
size = len(msg.get("content", ""))
total_size += size
if PERSISTED_OUTPUT_TAG not in msg.get("content", ""):
candidates.append((i, size))
def enforce_turn_budget(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."""
sizes = [len(msg.get("content", "")) for msg in tool_messages]
total_size = sum(sizes)
candidates = [(i, size) for i, size in enumerate(sizes)
if PERSISTED_OUTPUT_TAG not in tool_messages[i].get("content", "")]
if total_size <= config.turn_budget:
return tool_messages
candidates.sort(key=lambda x: x[1], reverse=True)
for idx, size in candidates:
for idx, size in sorted(candidates, key=lambda x: x[1], reverse=True):
if total_size <= config.turn_budget:
break
content = tool_messages[idx]["content"]
tools/tool_search.py
+118 -209
View File
@@ -1,14 +1,14 @@
"""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)."""
"""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``."""
from __future__ import annotations
import functools
import json
import logging
import math
@@ -17,40 +17,29 @@ from dataclasses import dataclass
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,
)
from tools.tool_search_catalog import ( # noqa: F401 — re-exported public/test names
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,
)
BRIDGE_TOOL_NAMES, CHARS_PER_TOKEN, TOOL_CALL_NAME, TOOL_DESCRIBE_NAME, TOOL_SEARCH_NAME,
CatalogEntry, _corpus_stats, _entry_search_text, _fn, _listing_group_label,
_registry_entry, _registry_toolset, _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 bridge call can request.
_MAX_QUERIES_PER_CALL = 10
_MAX_DESCRIBE_NAMES_PER_CALL = 10
_MAX_QUERIES_PER_CALL = _MAX_DESCRIBE_NAMES_PER_CALL = 10 # bound the work one bridge call requests
@dataclass(frozen=True)
class ToolSearchConfig:
"""Resolved, validated tool-search configuration for a single assembly."""
enabled: str # "auto" | "on" | "off" — "auto" is an alias of "on" today
# Listing budget as % of context. Does NOT gate activation; bounds how much
# the embedded listing may consume before it degrades (full -> names -> bare).
# Listing budget as % of context; does NOT gate activation, only bounds how much
# the embedded manifest may consume before it degrades (full -> names -> bare).
threshold_pct: float # 0..100
search_default_limit: int
max_search_limit: int
# 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
# Core/GUI names deferred behind the bridge. None = curated default; an
# explicit config list replaces it wholesale ([] = defer no core tools).
listing: str = "auto" # "auto"/"on" = embed the manifest when it fits; "off" = bare bridge
listing_max_tokens: int = 4000 # budget = min(this, threshold_pct% of context)
# None = curated default; an explicit list replaces it wholesale ([] = defer no core tools).
defer_tools: Optional[frozenset] = None
@property
@@ -59,11 +48,10 @@ class ToolSearchConfig:
@classmethod
def from_raw(cls, raw: Any) -> "ToolSearchConfig":
"""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)
"""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): # legacy bool / None
raw = {"enabled": "off" if raw is False else "auto"}
max_search_limit = _clamped_int(raw.get("max_search_limit"), 25, 1, 50)
defer_raw = raw.get("defer")
return cls(
@@ -74,10 +62,8 @@ class ToolSearchConfig:
max_search_limit=max_search_limit,
listing=_tri_state(raw.get("listing", "auto")),
listing_max_tokens=_clamped_int(raw.get("listing_max_tokens"), 4000, 200, 60000),
# 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))
defer_tools=(frozenset(str(n).strip() for n in defer_raw if str(n).strip())
if isinstance(defer_raw, (list, tuple, set)) else None))
_TRI_STATE_ALIASES = {"true": "on", "1": "on", "yes": "on", "false": "off", "0": "off", "no": "off"}
@@ -86,8 +72,7 @@ _TRI_STATE_ALIASES = {"true": "on", "1": "on", "yes": "on", "false": "off", "0":
def _tri_state(value: Any) -> str:
"""Normalize an ``auto``/``on``/``off`` setting (bool-ish aliases accepted)."""
text = str(value).strip().lower()
text = _TRI_STATE_ALIASES.get(text, text)
return text if text in ("auto", "on", "off") else "auto"
return _TRI_STATE_ALIASES.get(text, text if text in ("auto", "on", "off") else "auto")
def _clamped_int(value: Any, fallback: int, lo: int, hi: int) -> int:
@@ -110,26 +95,20 @@ 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 {}
tools_cfg = cfg.get("tools") if isinstance(cfg.get("tools"), dict) else {}
tools_cfg = (getattr(_cfg_mod, loader_name)() or {}).get("tools")
tools_cfg = tools_cfg if isinstance(tools_cfg, dict) else {}
return ToolSearchConfig.from_raw(tools_cfg.get("tool_search"))
except Exception as e:
logger.debug("Failed to load tool-search config: %s", e)
return ToolSearchConfig.from_raw(None)
def load_config() -> ToolSearchConfig:
return _config_from_loader("load_config")
def load_config_readonly() -> ToolSearchConfig:
"""Same as ``load_config`` without copying the cached full config."""
return _config_from_loader("load_config_readonly")
load_config = functools.partial(_config_from_loader, "load_config")
load_config_readonly = functools.partial(_config_from_loader, "load_config_readonly") # no copy
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)."""
"""Names that never defer by default (lazy: ``toolsets`` imports ``tools.registry``)."""
try:
from toolsets import _HERMES_CORE_TOOLS
return frozenset(_HERMES_CORE_TOOLS)
@@ -137,43 +116,36 @@ def _core_tool_names() -> frozenset[str]:
return frozenset()
# Session-gated GUI toolsets: off ``_HERMES_CORE_TOOLS`` so non-GUI clients never
# pay their schema; once enabled they stay direct unless the deferral list names them.
# Session-gated GUI toolsets: off ``_HERMES_CORE_TOOLS`` so non-GUI clients never pay
# their schema; once enabled they stay direct unless the deferral list names them.
_DIRECT_SURFACE_TOOLSETS = frozenset({"desktop_ui", "project"})
# Curated event-triggered core tools deferred BY DEFAULT — reached for when
# 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.
# Event-triggered core tools deferred BY DEFAULT (a catalog stub suffices); the ``defer``
# config replaces this wholesale ([] = everything eager). POST-rename names. ``clarify``
# is deliberately absent: A/B showed deferring it collapsed structured-clarify usage
# (18/18 -> 7/18) — the ask-the-user affordance must be ambient, a stub is not enough.
_DEFAULT_DEFERRED_TOOLS = frozenset({
"computer_use", "session_search", "image_generate",
"todo_list", "process_manage", "cronjob_manage",
# Desktop GUI surface (desktop_ui + project toolsets)
"drive_preview", "gui_tour", "desktop_preview", "annotate_preview",
"show_tip", "setup_mcp", "desktop_project", "close_terminal",
"apply_layout", "read_terminal", "read_window_below", "focus_pane",
})
"apply_layout", "read_terminal", "read_window_below", "focus_pane"})
def is_deferrable_tool_name(name: str, defer_tools: Optional[frozenset] = None) -> bool:
"""True if a tool is *eligible* for deferral: named in ``defer_tools``
(curated core set or user override), OR an MCP tool, OR neither core nor a
session-gated GUI surface (i.e. a plugin tool). Bridge names never defer."""
"""True if a tool is *eligible* for deferral: named in ``defer_tools`` (curated set or
user override), OR an MCP tool, OR neither core nor a session-gated GUI surface (i.e. a
plugin tool). Bridge names never defer."""
if name in BRIDGE_TOOL_NAMES:
return False
if defer_tools is not None and name in defer_tools:
return True
if name in _core_tool_names():
return False
entry = _registry_entry(name)
if entry is None:
return False
try:
return entry.toolset.startswith("mcp-") or entry.toolset not in _DIRECT_SURFACE_TOOLSETS
except Exception: # malformed entry (no str toolset) is never deferrable
return False
toolset = _registry_toolset(name) # None (unregistered/malformed) never defers
return toolset is not None and (
toolset.startswith("mcp-") or toolset not in _DIRECT_SURFACE_TOOLSETS)
def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
@@ -181,74 +153,59 @@ def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]:
return (_fn(td).get("name", "") for td in tool_defs)
def classify_tools(
tool_defs: 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)."""
def classify_tools(tool_defs: 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 (re-added
after classification)."""
visible: List[Dict[str, Any]] = []
deferrable: List[Dict[str, Any]] = []
for td, name in zip(tool_defs, _tool_def_names(tool_defs)):
if name in BRIDGE_TOOL_NAMES:
continue
(deferrable if is_deferrable_tool_name(name, defer_tools) else visible).append(td)
if name not in BRIDGE_TOOL_NAMES:
(deferrable if is_deferrable_tool_name(name, defer_tools) else visible).append(td)
return visible, deferrable
def _deferrable_in(tool_defs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Deferrable subset of a pre-assembly ``tool_defs`` list under the current
(read-only) user config — the universe the bridge tools operate on."""
"""Deferrable subset of pre-assembly ``tool_defs`` under the read-only user config."""
return classify_tools(tool_defs, load_config_readonly().effective_defer_tools)[1]
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)."""
total_chars = 0
for td in tool_defs:
"""Token cost via the chars/4 rule (order-of-magnitude precision suffices)."""
def _chars(td: Dict[str, Any]) -> int:
try:
total_chars += len(json.dumps(td, ensure_ascii=False, separators=(",", ":")))
return len(json.dumps(td, ensure_ascii=False, separators=(",", ":")))
except (TypeError, ValueError):
total_chars += len(str(td))
return int(math.ceil(total_chars / CHARS_PER_TOKEN))
return len(str(td))
return int(math.ceil(sum(map(_chars, tool_defs)) / CHARS_PER_TOKEN))
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 reserved for a future budget-gated mode — do not distinguish
them without that design). ``context_length`` is kept for caller compatibility."""
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 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:
"""``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:
pct_leg = int(context_length * (config.threshold_pct / 100.0))
else:
pct_leg = 10_000
"""``min(listing_max_tokens, threshold_pct% of context)``; unknown context uses a 10K
percentage leg (5% of a typical 200K window)."""
pct_leg = (int(context_length * (config.threshold_pct / 100.0))
if context_length and context_length > 0 else 10_000)
return max(0, min(config.listing_max_tokens, pct_leg))
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)."""
return {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": {"type": "object", "properties": properties, "required": required},
},
}
return {"type": "function", "function": {
"name": name, "description": description,
"parameters": {"type": "object", "properties": properties, "required": required}}}
def _search_description(deferred_count: int, listing: Optional[str], listing_form: str) -> str:
"""tool_search bridge description with the listing embedded; ``listing_form``
picks the framing (see :func:`bridge_tool_schemas`)."""
"""tool_search bridge description with the listing embedded (framing per ``listing_form``)."""
desc = (
f"Search {deferred_count} additional tools that are loaded on demand. "
"Takes a list of queries searched in parallel against the same "
@@ -280,15 +237,11 @@ def _search_description(deferred_count: int, listing: Optional[str], listing_for
return desc + "\n\n" + listing
def bridge_tool_schemas(
deferred_count: int,
listing: Optional[str] = None,
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", "groups" says which domains exist and
that search is mandatory)."""
def bridge_tool_schemas(deferred_count: int, listing: Optional[str] = None,
listing_form: str = "") -> List[Dict[str, Any]]:
"""Bridge tool schemas injected in place of deferred tools; kept short — every byte is paid
every turn. ``listing`` is embedded in the tool_search description; per-tool forms say
"skip search when you see the exact name", "groups" says search is mandatory."""
return [
_bridge_schema(
TOOL_SEARCH_NAME,
@@ -340,41 +293,32 @@ def bridge_tool_schemas(
@dataclass
class AssemblyResult:
"""Outcome of one assembly (tests and observability)."""
tool_defs: List[Dict[str, Any]]
activated: bool
deferred_count: int = 0
deferred_tokens: int = 0
threshold_tokens: int = 0
# 0 = passthrough; 1 = bridge + per-tool listing (full/names/mixed);
# 2 = bare bridge / server-summary only (tools reachable only via tool_search).
# 0 = passthrough; 1 = bridge + per-tool listing; 2 = bare bridge / server summary only.
tier: int = 0
listing_form: str = "none" # "full" | "names" | "mixed" | "groups" | "none"
def assemble_tool_defs(
tool_defs: List[Dict[str, Any]],
*,
context_length: Optional[int] = None,
config: Optional[ToolSearchConfig] = None) -> AssemblyResult:
def assemble_tool_defs(tool_defs: List[Dict[str, Any]], *, context_length: Optional[int] = None,
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()
config = config or 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)
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),
deferred_tokens=deferrable_tokens,
threshold_tokens=int((context_length or 0) * (config.threshold_pct / 100.0)),
tier=0)
threshold_tokens=int((context_length or 0) * (config.threshold_pct / 100.0)), tier=0)
listing, listing_form = None, "none"
listing_budget = listing_token_budget(config, context_length)
if config.listing != "off":
@@ -399,34 +343,29 @@ def is_bridge_tool(name: str) -> bool:
def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]:
"""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
required = params.get("required") if isinstance(params, dict) else None
if not isinstance(required, list):
try:
required = entry.schema["function"]["parameters"]["required"]
except (TypeError, KeyError, AttributeError):
required = []
return {
"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]}
return {"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(required, list) else [])
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."""
"""Deterministic ``[{name, tool_count}]`` of connected sources (attached to 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)]
def _string_list_arg(
args: Dict[str, Any], key: str, *, dedupe: bool, max_items: int, retry_hint: str,
) -> Tuple[Optional[List[str]], Optional[str]]:
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) is a one-item list; rejects non-lists, all-blank lists, > ``max_items``."""
raw = args.get(key)
if isinstance(raw, str):
raw = [raw]
raw = [raw] if isinstance(raw, str) else raw
if not isinstance(raw, list):
return None, tool_error(f"{key} is required and must be an array of strings")
out: List[str] = []
@@ -435,38 +374,26 @@ def _string_list_arg(
if text and (not dedupe or text not in out):
out.append(text)
if not out:
return None, tool_error(
f"{key} is required and must contain at least one non-empty string")
return None, tool_error(f"{key} is required and must contain at least one non-empty string")
if len(out) > max_items:
return None, tool_error(
f"too many {key}: {len(out)} > max {max_items}. {retry_hint}")
return None, tool_error(f"too many {key}: {len(out)} > max {max_items}. {retry_hint}")
return out, None
def dispatch_tool_search(args: Dict[str, Any],
*,
current_tool_defs: List[Dict[str, Any]],
def dispatch_tool_search(args: Dict[str, Any], *, current_tool_defs: List[Dict[str, Any]],
config: Optional[ToolSearchConfig] = None) -> str:
"""Execute the ``tool_search`` bridge tool. Returns JSON::
{"queries": [...], "total_available": N,
"results": [{"query": ..., "matches": [names...]}, ...],
"tools": {name: {"source", "source_name", "description", "required"}}}
``limit`` applies PER QUERY. Empty query groups get ``available_sources`` +
``hint`` so a lexical miss is not mistaken for a missing capability.
"""
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.")
"""Execute the ``tool_search`` bridge tool -> JSON ``{queries, total_available,
results: [{query, matches: [names]}], tools: {name: {source, source_name, description,
required}}}``. ``limit`` applies PER QUERY; empty groups get ``available_sources`` +
``hint`` so a lexical miss is not mistaken for a missing capability."""
config = config or 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")
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]] = {}
@@ -485,34 +412,21 @@ def dispatch_tool_search(args: Dict[str, Any],
"tool_search with the service name plus a concrete action or "
"object before concluding the capability is unavailable.")
results.append(group)
return json.dumps({
"queries": queries,
"total_available": len(catalog),
"results": results,
"tools": tools_map,
}, ensure_ascii=False)
return json.dumps({"queries": queries, "total_available": len(catalog), "results": results,
"tools": tools_map}, ensure_ascii=False)
def dispatch_tool_describe(args: Dict[str, Any],
*,
current_tool_defs: List[Dict[str, Any]],
def dispatch_tool_describe(args: Dict[str, Any], *, current_tool_defs: List[Dict[str, Any]],
config: Optional[ToolSearchConfig] = None) -> str:
"""Execute the ``tool_describe`` bridge tool. Returns JSON::
{"tools": {name: {"description", "parameters"}},
"not_found": [...], # unknown / not in this assembly (never fails the call)
"errors": {name: msg}} # registered but non-deferrable names
Duplicates are deduped silently.
"""
if config is None:
config = load_config_readonly()
"""Execute the ``tool_describe`` bridge tool -> JSON ``{tools: {name: {description,
parameters}}, not_found: [...] (unknown / not in this assembly; never fails the call),
errors: {name: msg} (registered but non-deferrable)}``. Duplicates dedupe silently."""
config = config or 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.")
if err:
return err
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]] = {}
@@ -521,9 +435,8 @@ def dispatch_tool_describe(args: Dict[str, Any],
for name in names:
fn = by_name.get(name)
if fn is not None:
tools[name] = {
"description": fn.get("description", ""),
"parameters": fn.get("parameters", {})}
tools[name] = {"description": fn.get("description", ""),
"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.
@@ -532,7 +445,6 @@ def dispatch_tool_describe(args: Dict[str, Any],
"already, call it directly; otherwise check the spelling against tool_search.")
else:
not_found.append(name)
result: Dict[str, Any] = {"tools": tools}
if not_found:
result["not_found"] = not_found
@@ -547,9 +459,8 @@ def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
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))
return frozenset(n for n in _tool_def_names(tool_defs)
if n and is_deferrable_tool_name(n, defer_tools))
def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[str, Any], Optional[str]]:
@@ -561,13 +472,12 @@ def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[s
if name in BRIDGE_TOOL_NAMES:
return None, {}, f"tool_call cannot invoke '{name}' (it is itself a bridge tool)"
raw_args = args.get("arguments")
if raw_args is None:
raw_args = {}
if isinstance(raw_args, str):
try:
raw_args = json.loads(raw_args)
except json.JSONDecodeError as e:
return None, {}, f"tool_call 'arguments' is not valid JSON: {e}"
raw_args = {} if raw_args is None else raw_args
if not isinstance(raw_args, dict):
return None, {}, "tool_call 'arguments' must be an object"
if not is_deferrable_tool_name(name, load_config_readonly().effective_defer_tools):
@@ -579,10 +489,9 @@ def resolve_underlying_call(args: Dict[str, Any]) -> Tuple[Optional[str], Dict[s
__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_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",
]
"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
+78 -124
View File
@@ -13,10 +13,13 @@ from typing import Any, Dict, List, Optional, Tuple
import snowballstemmer
from tools.tool_search_names import TOOL_CALL_NAME, TOOL_DESCRIBE_NAME, TOOL_SEARCH_NAME
# Chars-per-token rule of thumb for budget estimates; 4.0 slightly
# underestimates, which is the safer direction (fewer false activations).
# Reserved bridge names: a user/plugin/MCP tool may not take them (registry override
# protection rejects such registrations).
TOOL_SEARCH_NAME = "tool_search"
TOOL_DESCRIBE_NAME = "tool_describe"
TOOL_CALL_NAME = "tool_call"
BRIDGE_TOOL_NAMES = frozenset({TOOL_SEARCH_NAME, TOOL_DESCRIBE_NAME, TOOL_CALL_NAME})
# Chars-per-token rule of thumb; 4.0 slightly underestimates (fewer false activations).
CHARS_PER_TOKEN = 4.0
@@ -26,38 +29,30 @@ class CatalogEntry:
name: str
description: str
schema: Dict[str, Any] # The full {"type":"function", "function": {...}} entry.
schema: Dict[str, Any] # the full {"type":"function", "function": {...}} entry
source: str # "mcp" | "plugin" | "other"
source_name: str # Toolset name, e.g. "mcp-github" or "kanban"
source_name: str # toolset name, e.g. "mcp-github" or "kanban"
_tokens: List[str] = field(default_factory=list) # pre-tokenized for BM25
_TOKEN_RE = re.compile(r"[A-Za-z0-9]+")
# Snowball stemmers carry mutable parsing state and bridge dispatch runs on
# parallel tool-call threads, so: one stemmer per thread, created lazily.
_thread_local = threading.local()
def _stemmer() -> Any:
st = getattr(_thread_local, "stemmer", None)
if st is None:
st = _thread_local.stemmer = snowballstemmer.stemmer("english")
return st
@functools.lru_cache(maxsize=16384)
def _stem(token: str) -> str:
"""Stem one token, memoized across stateless catalog rebuilds."""
return _stemmer().stemWord(token)
"""Stem one token, memoized across stateless catalog rebuilds. Snowball stemmers carry
mutable parsing state and bridge dispatch runs on parallel tool-call threads, so the
stemmer is one-per-thread, created lazily."""
if getattr(_thread_local, "stemmer", None) is None:
_thread_local.stemmer = snowballstemmer.stemmer("english")
return _thread_local.stemmer.stemWord(token)
def _tokenize(text: str) -> List[str]:
"""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)]
"""Lowercase alphanumeric tokens, Snowball-stemmed (English); shared by the index and
query paths so "issues" matches ``create_issue``."""
return [_stem(token.lower()) for token in _TOKEN_RE.findall(text)] if text else []
def _fn(td: Dict[str, Any]) -> Dict[str, Any]:
@@ -66,9 +61,8 @@ def _fn(td: Dict[str, Any]) -> Dict[str, Any]:
def _registry_entry(name: str) -> Any:
"""Registry entry for ``name``; None when unregistered OR when the registry
is unavailable/raises (lookup failures must never fail a bridge call).
The import stays lazy: tests patch ``tools.registry.registry``."""
"""Registry entry for ``name``; None when unregistered OR the registry raises (lookup
failures must never fail a bridge call). Lazy import: tests patch the registry."""
try:
from tools.registry import registry
return registry.get_entry(name)
@@ -76,33 +70,33 @@ def _registry_entry(name: str) -> Any:
return None
def _registry_toolset(name: str) -> Optional[str]:
"""Toolset of a registered tool; None when unregistered or malformed (no str toolset)."""
toolset = getattr(_registry_entry(name), "toolset", None)
return toolset if isinstance(toolset, str) else None
def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str:
"""Search-text blob: split name words + source label + description +
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."""
"""Search-text blob: split name words + source label + description + top-level parameter
names (schema bodies are noise with 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."""
fn = _fn(td)
name = fn.get("name", "")
if name.startswith("mcp__"):
name = name[len("mcp__"):]
desc = fn.get("description", "") or ""
params = ((fn.get("parameters") or {}).get("properties") or {})
param_names = " ".join(params.keys())
name_words = re.sub(r"[_.:-]", " ", name)
extra = source_label if source_label and source_label not in name_words.split() else ""
return f"{name_words} {extra} {desc} {param_names}"
param_names = " ".join(((fn.get("parameters") or {}).get("properties") or {}).keys())
return f"{name_words} {extra} {fn.get('description', '') or ''} {param_names}"
def _classify_source(name: str) -> Tuple[str, str]:
"""Return (source_kind, source_name) for a registered tool name."""
entry = _registry_entry(name)
if entry is None:
return ("other", "")
try:
return ("mcp" if entry.toolset.startswith("mcp-") else "plugin", entry.toolset)
except Exception: # malformed entry (no str toolset)
toolset = _registry_toolset(name)
if toolset is None:
return ("other", "")
return ("mcp" if toolset.startswith("mcp-") else "plugin", toolset)
def build_catalog(tool_defs: List[Dict[str, Any]]) -> List[CatalogEntry]:
@@ -117,80 +111,57 @@ def build_catalog(tool_defs: List[Dict[str, Any]]) -> List[CatalogEntry]:
# Index the human-facing label ("linear", not "mcp-linear").
source_label = _listing_group_label(source_name) if source_name else ""
catalog.append(CatalogEntry(
name=name,
description=fn.get("description", "") or "",
schema=td,
source=source,
source_name=source_name,
_tokens=_tokenize(_entry_search_text(td, source_label))))
name=name, description=fn.get("description", "") or "", schema=td, source=source,
source_name=source_name, _tokens=_tokenize(_entry_search_text(td, source_label))))
return catalog
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 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
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 for one query against one document (inlined; the catalog is bounded —
typically < 500 tools — so a dependency is not worth it)."""
score = 0.0
dl = len(doc_tokens)
doc_tf = Counter(doc_tokens)
for q in query_tokens:
df = doc_freq.get(q, 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))
norm = tf * (k1 + 1) / (tf + k1 * (1 - b + b * dl / max(avg_dl, 1.0)))
score += idf * norm
df, tf = doc_freq.get(q, 0), doc_tf.get(q, 0)
if df and tf:
idf = math.log(1 + (n_docs - df + 0.5) / (df + 0.5))
score += idf * tf * (k1 + 1) / (tf + k1 * (1 - b + b * dl / max(avg_dl, 1.0)))
return score
_CorpusStats = Tuple[List[int], float, Dict[str, int], int]
_CorpusStats = Tuple[List[int], float, Dict[str, int], int] # doc_lengths, avg_dl, df, n_docs
def _corpus_stats(catalog: List[CatalogEntry]) -> _CorpusStats:
"""Compute the BM25 statistics shared by every query over a catalog."""
doc_lengths = [len(entry._tokens) for entry in catalog]
avg_dl = sum(doc_lengths) / max(len(doc_lengths), 1)
doc_freq: Dict[str, int] = Counter()
for entry in catalog:
doc_freq.update(set(entry._tokens))
doc_freq = Counter(tok for entry in catalog for tok in set(entry._tokens))
return doc_lengths, avg_dl, dict(doc_freq), len(catalog)
def search_catalog(
catalog: List[CatalogEntry],
query: str,
limit: int = 5,
*,
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."""
if not catalog or limit <= 0:
return []
query_tokens = _tokenize(query)
def search_catalog(catalog: List[CatalogEntry], query: str, limit: int = 5, *,
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."""
query_tokens = _tokenize(query) if catalog and limit > 0 else []
if not query_tokens:
return []
if corpus_stats is None:
corpus_stats = _corpus_stats(catalog)
corpus_stats = corpus_stats or _corpus_stats(catalog)
scored: List[Tuple[float, CatalogEntry]] = []
exact_name = query.strip().lower()
for entry in catalog:
if entry.name.lower() == exact_name:
scored.append((float("inf"), entry))
continue
s = _bm25_score(query_tokens, entry._tokens, *corpus_stats)
s = (float("inf") if entry.name.lower() == exact_name
else _bm25_score(query_tokens, entry._tokens, *corpus_stats))
if s > 0:
scored.append((s, entry))
if not scored:
ql = query.lower()
scored = [(0.1, entry) for entry in catalog if ql in entry.name.lower()]
scored = [(0.1, entry) for entry in catalog if query.lower() in entry.name.lower()]
scored.sort(key=lambda x: x[0], reverse=True)
return [e for _, e in scored[:limit]]
@@ -200,20 +171,16 @@ _SENTENCE_END_RE = re.compile(r"(?<!\be\.g)(?<!\bi\.e)(?<!\betc)[.!?](?=\s|$)")
def _short_desc(description: str, max_chars: int = 60) -> str:
"""First sentence of a tool description, clipped to ``max_chars`` on a
word boundary. ``e.g.``/``i.e.``/``etc.`` do not end a sentence; whitespace
normalization and the regex search stay linear-time on hostile input."""
"""First sentence of a tool description, clipped to ``max_chars`` on a word boundary.
``e.g.``/``i.e.``/``etc.`` do not end a sentence; whitespace normalization and the regex
search stay linear-time on hostile input."""
text = " ".join((description or "").split())
if not text:
return ""
m = _SENTENCE_END_RE.search(text)
if m:
text = text[:m.end()]
text = text[:m.end()] if m else text
if len(text) <= max_chars:
return text
clipped = text[:max_chars]
if " " in clipped:
clipped = clipped.rsplit(" ", 1)[0]
clipped = clipped.rsplit(" ", 1)[0] if " " in clipped else clipped
return clipped.rstrip(",;: ") + "…"
@@ -225,29 +192,20 @@ def _listing_group_label(source_name: str) -> str:
def build_catalog_listing_with_form(
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).
Returns ``(text, form)``; ``form`` is ``"full"``, ``"names"`` (names-only),
``"mixed"`` (oversized servers collapsed to a name + count summary line,
small ones keep per-tool lines), ``"groups"`` (every server summarized),
or ``"none"`` (over budget even summarized -> text is None).
Ordering is deterministic (sorted groups and tools) so the block is
byte-stable across assemblies — the request prefix stays cacheable.
Degradation is PER SERVER (largest first): one huge server must not cost
a small co-attached server its listing.
"""
"""Render the deferred-catalog manifest: ``- name: short desc`` lines grouped per source.
Returns ``(text, form)``; form is ``"full"``, ``"names"``, ``"mixed"`` (oversized servers
collapsed to a name + count line), ``"groups"`` (every server summarized) or ``"none"``
(over budget even summarized -> text is None). Ordering is deterministic (sorted groups
and tools) so the block is byte-stable — the request prefix stays cacheable. Degradation
is PER SERVER, largest first: one huge server must not cost a small one its listing."""
groups: Dict[str, List[Tuple[str, str]]] = {}
for td in deferrable:
fn = _fn(td)
name = fn.get("name", "")
if not name:
continue
# ``_classify_source`` returns ("other", "") for unregistered names and
# ``_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 name:
# _classify_source gives ("other", "") when unregistered; the label of "" is "other".
label = _listing_group_label(_classify_source(name)[1])
groups.setdefault(label, []).append((name, _short_desc(fn.get("description", ""))))
if not groups:
return None, "none"
@@ -271,19 +229,15 @@ def build_catalog_listing_with_form(
text = "\n".join([header] + [render_group(lbl, modes[lbl]) for lbl in sorted(groups)])
return text if math.ceil(len(text) / CHARS_PER_TOKEN) <= max_tokens else None
# 1. Everything full. 2. Everything names-only.
for mode in ("full", "names"):
for mode in ("full", "names"): # 1. everything full; 2. everything names-only
modes = {lbl: mode for lbl in groups}
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).
by_size = sorted(groups, key=lambda lbl: (-len(render_group(lbl, "names")), lbl))
for lbl in by_size:
# 3. Collapse the LARGEST rendered groups first (deterministic: size then label).
for lbl in sorted(groups, key=lambda lbl: (-len(render_group(lbl, "names")), lbl)):
modes[lbl] = "summary"
text = assemble_if_fits(modes)
if text is not None:
form = "groups" if all(m == "summary" for m in modes.values()) else "mixed"
return text, form
return text, "groups" if all(m == "summary" for m in modes.values()) else "mixed"
return None, "none"
tools/tool_search_names.py
-9
View File
@@ -1,9 +0,0 @@
"""Reserved bridge tool names shared by tool_search and its catalog module."""
# Reserved: a user/plugin/MCP tool may not take these names — the registry's
# override protection rejects such registrations.
TOOL_SEARCH_NAME = "tool_search"
TOOL_DESCRIBE_NAME = "tool_describe"
TOOL_CALL_NAME = "tool_call"
BRIDGE_TOOL_NAMES = frozenset({TOOL_SEARCH_NAME, TOOL_DESCRIBE_NAME, TOOL_CALL_NAME})
tools/tool_search_validation.py
+29 -44
View File
@@ -16,46 +16,40 @@ _SCHEMA_LITERAL_KEYS = frozenset({"const", "default", "enum", "example", "exampl
def _schema_for_local_validation(node: Any) -> Any:
"""JSON-Schema-compatible copy honoring OpenAPI ``nullable: true`` — the
normal coercion path accepts that shape, so local validation must too."""
"""JSON-Schema-compatible copy honoring OpenAPI ``nullable: true`` (the normal coercion
path accepts that shape, so local validation must too)."""
if isinstance(node, list):
return [_schema_for_local_validation(item) for item in node]
if not isinstance(node, dict):
return node
# 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"}
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":
normalized["type"] = [schema_type, "null"]
return normalized
schema_type = [schema_type]
if isinstance(schema_type, list):
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.
# 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"}]}
def _schema_has_external_ref(node: Any) -> bool:
"""True when *node* contains a non-local ``$ref`` — local validation must
never turn a tool call into an implicit network/file fetch (fail open)."""
"""True when *node* contains a non-local ``$ref`` — local validation must never turn a
tool call into an implicit network/file fetch (fail open)."""
if isinstance(node, list):
return any(_schema_has_external_ref(item) for item in node)
if not isinstance(node, dict):
return False
ref = node.get("$ref")
if isinstance(ref, str) and not ref.startswith("#"):
return True
return any(
_schema_has_external_ref(value)
for key, value in node.items()
return (isinstance(ref, str) and not ref.startswith("#")) or any(
_schema_has_external_ref(value) for key, value in node.items()
if key not in _SCHEMA_LITERAL_KEYS)
@@ -63,12 +57,10 @@ def _validation_path(error: Any) -> str:
"""Format a jsonschema error path as a compact argument path."""
path = "arguments"
for part in getattr(error, "absolute_path", ()):
if isinstance(part, int):
path += f"[{part}]"
elif isinstance(part, str) and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", part):
if isinstance(part, str) and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", part):
path += f".{part}"
else:
path += f"[{json.dumps(part, ensure_ascii=False)}]"
path += f"[{part if isinstance(part, int) else json.dumps(part, ensure_ascii=False)}]"
return path
@@ -79,15 +71,12 @@ def _validation_error(message: str, *, path: str, constraint: str, parameters: A
def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str]:
"""Validate ``tool_call`` arguments against the deferred tool's schema.
Models invoke deferred tools "blind" (schema unseen) and omit required
args; without this, the opaque downstream failure makes cheap models loop.
Required-field probe first, then the same schema-guided coercion normal
dispatch applies, then jsonschema on the repaired copy. Missing/malformed
schemas, no validator, and external refs all fail OPEN. Returns a JSON
error string when invalid, ``None`` when the call should dispatch.
"""
"""Validate ``tool_call`` arguments against the deferred tool's schema. Models invoke
deferred tools "blind" (schema unseen) and omit required args; without this, the opaque
downstream failure makes cheap models loop. Required-field probe first, then the same
schema-guided coercion normal dispatch applies, then jsonschema on the repaired copy.
Missing/malformed schemas, no validator, and external refs all fail OPEN. Returns a JSON
error string when invalid, ``None`` when the call should dispatch."""
try:
from tools.registry import registry as _registry
schema = _registry.get_schema(name)
@@ -98,35 +87,31 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str
if not isinstance(params, dict):
return None
required = params.get("required")
if isinstance(required, list):
missing = [r for r in required if isinstance(r, str) and r not in args]
if missing:
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)
missing = ([r for r in required if isinstance(r, str) and r not in args]
if isinstance(required, list) else [])
if missing:
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)
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)
return None
# Validate the repaired shape dispatch will see; copy because
# coerce_tool_args may normalize in place (dispatch re-coerces canonically).
# Validate the repaired shape dispatch will see; copy because coerce_tool_args may
# normalize in place (dispatch re-coerces canonically).
try:
from model_tools import coerce_tool_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)
try:
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)
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))
tools/tour_tool.py
+20 -48
View File
@@ -1,11 +1,9 @@
#!/usr/bin/env python3
"""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."""
"""Guided tour (highlight + narrate UI elements) in the Hermes desktop GUI: 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
@@ -18,36 +16,24 @@ SURFACES = ("app", "preview")
SIDES = ("top", "right", "bottom", "left")
def tour_tool(
action: str = "",
surface: Optional[str] = None,
selector: Optional[str] = None,
title: Optional[str] = None,
text: Optional[str] = None,
side: Optional[str] = None,
steps: Optional[list] = None,
step_index: Optional[int] = None,
callback: Optional[Callable] = None,
) -> str:
def tour_tool(action: str = "", surface: Optional[str] = None, selector: Optional[str] = None,
title: Optional[str] = None, text: Optional[str] = None, side: Optional[str] = None,
steps: Optional[list] = None, step_index: Optional[int] = None,
callback: Optional[Callable] = None) -> str:
"""Dispatch one tour action to the desktop renderer and return its outcome."""
if callback is None:
return tool_error("tour is only available in the Hermes desktop app.")
verb = (action or "").strip().lower()
if verb not in ACTIONS:
return tool_error(f"action must be one of: {', '.join(ACTIONS)}.")
where = (surface or "app").strip().lower()
if where not in SURFACES:
return tool_error(f"surface must be one of: {', '.join(SURFACES)}.")
if side is not None and side not in SIDES:
return tool_error(f"side must be one of: {', '.join(SIDES)}.")
# Every highlighted moment needs something to point at or something to say.
if verb == "show" and not (selector or title or text):
return tool_error("show needs a selector (and/or title/text for the popover).")
if verb == "start":
if not isinstance(steps, list) or not steps:
return tool_error("start needs a non-empty steps array.")
@@ -56,24 +42,15 @@ def tour_tool(
return tool_error(f"steps[{i}] must be an object.")
if not (step.get("selector") or step.get("title") or step.get("text")):
return tool_error(f"steps[{i}] needs a selector and/or title/text.")
fields = {
"action": verb, "surface": where, "selector": selector, "title": title,
"text": text, "side": side, "steps": steps, "step_index": step_index,
}
payload = {key: val for key, val in fields.items() if val is not None}
fields = {"action": verb, "surface": where, "selector": selector, "title": title,
"text": text, "side": side, "steps": steps, "step_index": step_index}
try:
raw = callback(payload)
raw = callback({key: val for key, val in fields.items() if val is not None})
except Exception as exc:
return tool_error(f"Tour action failed: {exc}")
if not raw:
return tool_error(
"The tour request timed out, or no GUI window answered. "
"For surface='preview' open a page in the preview pane first."
)
return tool_error("The tour request timed out, or no GUI window answered. "
"For surface='preview' open a page in the preview pane first.")
# The renderer answers with a JSON object; pass it through, else wrap it.
try:
return json.dumps(json.loads(raw), ensure_ascii=False)
@@ -159,14 +136,9 @@ def check_tours_enabled() -> bool:
registry.register(
name="gui_tour",
toolset="desktop_ui",
schema=TOUR_SCHEMA,
name="gui_tour", toolset="desktop_ui", schema=TOUR_SCHEMA, check_fn=check_tours_enabled,
handler=lambda args, **kw: tour_tool(
action=args.get("action", ""),
**{k: args.get(k) for k in ("surface", "selector", "title", "text", "side", "steps", "step_index")},
callback=kw.get("callback"),
),
check_fn=check_tours_enabled,
emoji="🧭",
)
action=args.get("action", ""), callback=kw.get("callback"),
**{k: args.get(k) for k in ("surface", "selector", "title", "text", "side", "steps",
"step_index")}),
emoji="🧭")