hermes-agent/tools/tool_search.py

"""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
from collections import Counter
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional, Tuple

from hermes_cli.config_defaults import DEFAULT_CONFIG
from tools.registry import tool_error
from tools.tool_search_catalog import (
    BRIDGE_TOOL_NAMES, CHARS_PER_TOKEN, TOOL_CALL_NAME, TOOL_DESCRIBE_NAME, TOOL_SEARCH_NAME,
    CatalogEntry, _fn, _listing_group_label, _registry_entry, _registry_toolset,
    build_catalog, build_catalog_listing_with_form, search_catalog)
from tools.tool_search_validation import (
    local_batch_error, normalize_tool_call_entries, not_deferrable_error, validate_deferred_call_args)
from tools.connectors import CONNECTOR_BATCH_SENTINEL, is_connector_name
from tools.connectors.search import (
    connections_in_scope, connector_entries_by_group, connectors_unavailable, remote_schemas_for)

logger = logging.getLogger("tools.tool_search")
# Bound the work one bridge call requests. Search is capped at the gateway's
# own limit: the connector search route answers 7 use_cases per request and
# returns HTTP 502 for 8 or more (measured 2026-09-09), and one local call
# maps to one gateway request. Describe has no such remote limit.
_MAX_QUERIES_PER_CALL = 7
_MAX_DESCRIBE_NAMES_PER_CALL = 10


@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, 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
    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
    def effective_defer_tools(self) -> frozenset:
        return _DEFAULT_DEFERRED_TOOLS if self.defer_tools is None else self.defer_tools

    @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):  # 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")
        if defer_raw is not None and not isinstance(defer_raw, (list, tuple, set)):
            # Loud, then the curated default: a scalar here means the user tried to shrink the
            # tool surface and got nothing — never silently ignore it (#116404).
            logger.warning(
                "tools.tool_search.defer is %r, expected a YAML list of tool names "
                "(e.g. [todo_list, computer_use]; [] keeps every tool eager) - "
                "using the curated default set.", defer_raw)
            defer_raw = None
        return cls(
            enabled=_tri_state(raw.get("enabled", "auto")),
            threshold_pct=max(0.0, min(100.0, _safe_float(raw.get("threshold_pct"), 5.0))),
            search_default_limit=_clamped_int(
                raw.get("search_default_limit"), 5, 1, max_search_limit),
            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),
            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"}


def _tri_state(value: Any) -> str:
    """Normalize an ``auto``/``on``/``off`` setting (bool-ish aliases accepted)."""
    text = str(value).strip().lower()
    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:
    """``int(value)`` (or ``fallback`` when unparseable) clamped to ``[lo, hi]``."""
    try:
        value = int(value)
    except (TypeError, ValueError):
        value = fallback
    return max(lo, min(hi, value))


def _safe_float(value: Any, fallback: float) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return fallback


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
        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)


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]:
    """Names that never defer by default (lazy: ``toolsets`` imports ``tools.registry``)."""
    try:
        from toolsets import _HERMES_CORE_TOOLS
        return frozenset(_HERMES_CORE_TOOLS)
    except Exception:
        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.
_DIRECT_SURFACE_TOOLSETS = frozenset({"desktop_ui", "project"})

# Event-triggered tools deferred BY DEFAULT (a catalog stub suffices). Keep the curated
# list in DEFAULT_CONFIG so config discovery and runtime behavior cannot drift. An explicit
# ``defer`` list replaces this wholesale ([] = everything eager). ``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(DEFAULT_CONFIG["tools"]["tool_search"]["defer"])


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 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
    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]:
    """Function names of a tool-defs list (``""`` for a nameless def)."""
    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 (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 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 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 via the chars/4 rule (order-of-magnitude precision suffices)."""
    def _chars(td: Dict[str, Any]) -> int:
        try:
            return len(json.dumps(td, ensure_ascii=False, separators=(",", ":")))
        except (TypeError, ValueError):
            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], *, connections_granted: bool = False) -> 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."""
    if config.enabled == "off":
        return False
    if deferrable_tokens > 0:
        return True
    return connections_granted


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)."""
    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}}}


_CONNECTIONS_HINT = (
    " Names starting with `connectors__` are tools of remote connector accounts "
    "(Gmail, Linear, Notion, ...); `manage_connections` checks whether an account "
    "is connected and gets the authorization link when it is not.")


def _search_description(deferred_count: int, listing: Optional[str], listing_form: str,
                        connections_granted: bool = False) -> str:
    """tool_search bridge description with the listing embedded (framing per ``listing_form``).
    ``connections_granted`` adds the one sentence that ties ``connectors__`` names to
    ``manage_connections``; without that tool in the session the sentence would name a tool
    the model cannot call."""
    desc = (
        (f"Search {deferred_count} additional tools that are loaded on demand. "
         if deferred_count else "Search remote connector tools (email, calendars, issue trackers, and more). ")
        + "Takes a list of queries searched in parallel against the same "
        "catalog; send one query per distinct capability you need. Returns "
        "matching tool names grouped per query plus a shared map with each "
        "tool's description. Follow with "
        f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, "
        f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this "
        "system prompt are already available and do not need to be searched."
        + (_CONNECTIONS_HINT if connections_granted else ""))
    if not listing:
        return desc
    if listing_form == "groups":
        return desc + (
            "\n\nThe servers below are connected and their tools ARE available "
            "through this bridge. For any request in these domains, search "
            "here FIRST — do not claim the capability is unavailable and do "
            "not substitute a generic tool (terminal/browser) without "
            "searching.\n\n" + listing)
    desc += (
        "\n\nEvery deferred capability is listed below. If a tool name "
        "appears here, do NOT claim it is unavailable — load it with "
        f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you "
        "already see the exact name).")
    if listing_form == "mixed":
        desc += (
            " For servers marked 'names not listed', the tools exist "
            f"too — find them with `{TOOL_SEARCH_NAME}` before "
            "concluding anything is missing.")
    return desc + "\n\n" + listing


def bridge_tool_schemas(deferred_count: int, listing: Optional[str] = None,
                        listing_form: str = "", connections_granted: bool = False) -> 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,
            _search_description(deferred_count, listing, listing_form, connections_granted),
            {
                "queries": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Search queries, each a few keywords describing one capability (e.g. ['create github issue', 'send slack message']). Searched in parallel; results come back grouped per query. A single string is accepted and treated as one query.",
                },
                "limit": {
                    "type": "integer",
                    "description": "Maximum number of matches per query. Defaults to 5 and is clamped to the configured maximum (25 by default).",
                },
            },
            ["queries"],
        ),
        _bridge_schema(
            TOOL_DESCRIBE_NAME,
            f"Load the full JSON schemas for tools returned by `{TOOL_SEARCH_NAME}`. "
            f"Required before `{TOOL_CALL_NAME}` if a tool's parameters are unknown. "
            "Batch every schema you need into one call.",
            {
                "names": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Exact tool names (as returned by tool_search). A single string is accepted and treated as one name.",
                },
            },
            ["names"],
        ),
        _bridge_schema(
            TOOL_CALL_NAME,
            "Invoke deferred tools. Takes `calls`, an array of {name, arguments} "
            "— one entry per invocation; a single call is an array of one. "
            "Local tools require one entry per tool_call. Only connectors__ names "
            "may be batched together; mixed and multi-local batches are rejected. "
            "Connector entries execute individually with results in input order. "
            f"Argument shapes match each tool's schema (see `{TOOL_DESCRIBE_NAME}`). "
            "Policy, hooks, and approvals run as for directly-listed tools.",
            {
                "calls": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "name": {"type": "string", "description": "Exact tool name to invoke."},
                            "arguments": {"type": "object", "description": "Arguments matching the tool schema."},
                        },
                        "required": ["name", "arguments"],
                    },
                    "description": "One local invocation, or one or more connector invocations. Never mix local and connector tools.",
                },
            },
            ["calls"],
        ),
    ]


@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; 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:
    """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."""
    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)
    connections_granted = connections_in_scope(incoming)
    if not deferrable:
        if should_activate(config, 0, context_length, connections_granted=connections_granted):
            return AssemblyResult(tool_defs=incoming + bridge_tool_schemas(0, connections_granted=connections_granted),
                                  activated=True, tier=2)
        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)
    listing, listing_form = None, "none"
    listing_budget = listing_token_budget(config, context_length)
    if config.listing != "off":
        listing, listing_form = build_catalog_listing_with_form(
            deferrable, max_tokens=listing_budget)
    bridge = bridge_tool_schemas(len(deferrable), listing=listing, listing_form=listing_form,
                                 connections_granted=connections_granted)
    tier = 1 if listing_form in ("full", "names", "mixed") else 2
    logger.info(
        "tool_search activated (tier %d): %d core/visible tools kept, %d deferred "
        "(~%d tokens), listing %s (budget ~%d tokens)",
        tier, len(visible), len(deferrable), deferrable_tokens, listing_form, listing_budget)
    return AssemblyResult(
        tool_defs=visible + bridge, activated=True, deferred_count=len(deferrable),
        deferred_tokens=deferrable_tokens, threshold_tokens=listing_budget,
        tier=tier, listing_form=listing_form)


def is_bridge_tool(name: str) -> bool:
    return name in BRIDGE_TOOL_NAMES


def _clip_description(text: str, cap: int = 500) -> str:
    """Cap a record description, marking the cut so it reads as deliberate.

    A bare slice ends mid-word ("apply exponential bac") and looks like
    corruption; the ellipsis says "there is more — tool_describe has it".
    500 keeps 9 in 10 vendor connector descriptions whole and every first
    sentence (measured p90 575, first-sentence max 329 over 353 tools).
    """
    return text if len(text) <= cap else text[:cap] + "…"


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."""
    try:
        required = entry.schema["function"]["parameters"]["required"]
    except (TypeError, KeyError, AttributeError):
        required = []
    return {"source": entry.source, "source_name": entry.source_name,
            "description": _clip_description(entry.description or ""),
            "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)."""
    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]]:
    """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)
    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] = []
    for item in raw:
        text = str(item or "").strip()
        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")
    if len(out) > max_items:
        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]],
                         config: Optional[ToolSearchConfig] = None,
                         connector_search: Optional[Any] = None) -> str:
    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))
    remote_entries: List[List[CatalogEntry]] = [[] for _ in queries]
    hosted_failure: Optional[str] = None
    if connections_in_scope(current_tool_defs):
        remote_entries, hosted_failure = connector_entries_by_group(
            queries, connector_search=connector_search)
    results: List[Dict[str, Any]] = []
    tools_map: Dict[str, Dict[str, Any]] = {}
    available_sources = _available_source_summary(catalog) if catalog else []
    for position, query in enumerate(queries):
        corpus = catalog + remote_entries[position]
        hits = search_catalog(corpus, query, limit=limit)
        for h in hits:
            tools_map.setdefault(h.name, _shared_tool_record(h))
        matches = [h.name for h in hits]
        group: Dict[str, Any] = {"query": query, "matches": matches}
        if not matches and catalog:
            group["available_sources"] = available_sources
            group["hint"] = (
                "This query returned no lexical matches, but the sources above "
                "are connected and their tools remain available. Retry "
                "tool_search with the service name plus a concrete action or "
                "object before concluding the capability is unavailable.")
        results.append(group)
    remote_count = sum(1 for name in tools_map if is_connector_name(name))
    payload: Dict[str, Any] = {"queries": queries, "total_available": len(catalog) + remote_count,
                               "results": results, "tools": tools_map}
    if hosted_failure:
        payload["connectors"] = connectors_unavailable(hosted_failure, verb="searched")
    return json.dumps(payload, ensure_ascii=False)


def dispatch_tool_describe(args: Dict[str, Any], *, current_tool_defs: List[Dict[str, Any]],
                           config: Optional[ToolSearchConfig] = None,
                           connector_describe: Optional[Any] = None) -> str:
    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}
    remote_schemas, hosted_failure = remote_schemas_for(names, current_tool_defs, connector_describe)

    tools: Dict[str, Dict[str, Any]] = {}
    not_found: List[str] = []
    undescribed: List[str] = []
    errors: Dict[str, str] = {}
    for name in names:
        fn = by_name.get(name)
        remote_fn = remote_schemas.get(name)
        if fn is not None:
            tools[name] = {"description": fn.get("description", ""),
                           "parameters": fn.get("parameters", {})}
        elif isinstance(remote_fn, dict):
            tools[name] = {"description": str(remote_fn.get("description", "")),
                           "parameters": remote_fn.get("parameters", {})}
        elif is_connector_name(name):
            (undescribed if hosted_failure else not_found).append(name)
        elif _registry_entry(name) is not None and not is_deferrable_tool_name(
            name, load_config_readonly().effective_defer_tools):
            # Registered but bridge/core/GUI-surface: a real name, wrong door.
            errors[name] = not_deferrable_error(name)
        else:
            not_found.append(name)
    result: Dict[str, Any] = {"tools": tools}
    if not_found:
        result["not_found"] = not_found
        result["hint"] = "Names in not_found are not currently available. Re-run tool_search to refresh."
    if errors:
        result["errors"] = errors
    if hosted_failure:
        result["connectors"] = connectors_unavailable(hosted_failure, verb="described", names=undescribed)
    return json.dumps(result, ensure_ascii=False)


def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]:
    """Deferrable names in the *pre-assembly* ``tool_defs`` of the session scope — the
    universe ``tool_call`` may reach. Gates bridge dispatch AND the executor unwrap so a
    restricted session cannot invoke an out-of-scope tool via the bridge."""
    defer_tools = load_config_readonly().effective_defer_tools
    return frozenset(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]]:
    """Parse a ``tool_call`` invocation into (underlying_name, args, error_msg).

    Used by:
    * the dispatcher in ``model_tools.handle_function_call``,
    * the display layer (so the activity feed shows the underlying tool),
    * the trajectory recorder.

    A connector-only batch resolves
    to ``(CONNECTOR_BATCH_SENTINEL, {"calls": [...]}, None)``: the batch is
    one dispatch unit owned by the ``model_tools`` bridge branch, and the
    sentinel is what planners/display layers see. A single local entry keeps
    the historical single-tool contract unchanged.

    On parse error, returns ``(None, {}, error_message)``.
    """
    entries, err = normalize_tool_call_entries(args)
    if err:
        return None, {}, err

    if len(entries) > 1 and any(not is_connector_name(e["name"]) for e in entries):
        return None, {}, local_batch_error(entries)
    if is_connector_name(entries[0]["name"]):
        return CONNECTOR_BATCH_SENTINEL, {"calls": entries}, None

    name = entries[0]["name"]
    raw_args = entries[0]["arguments"]
    if not is_deferrable_tool_name(name, load_config_readonly().effective_defer_tools):
        return None, {}, not_deferrable_error(name)
    return name, raw_args, None


__all__ = [
    "TOOL_SEARCH_NAME", "TOOL_DESCRIBE_NAME", "TOOL_CALL_NAME", "BRIDGE_TOOL_NAMES",
    "ToolSearchConfig", "CatalogEntry", "AssemblyResult", "load_config", "is_deferrable_tool_name",
    "classify_tools", "estimate_tokens_from_schemas", "should_activate", "build_catalog",
    "build_catalog_listing_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", "normalize_tool_call_entries",
    "CONNECTOR_BATCH_SENTINEL", "is_connector_name"]


# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
# The whole block is removed by reverting the commit that added it.
from typing import Literal  # noqa: F401,E402
import copy  # noqa: F401,E402
from dataclasses import field  # noqa: F401,E402
import re  # noqa: F401,E402
import snowballstemmer  # noqa: F401,E402
import threading  # noqa: F401,E402

def build_catalog_listing(
    deferrable: List[Dict[str, Any]],
    *,
    max_tokens: int = 4000,
) -> Optional[str]:
    """Render a skills-style manifest of the deferred catalog.

    One line per tool — ``name: short description`` — grouped under a
    heading per source (MCP server / plugin toolset), exactly like the
    bundled-skills listing in the system prompt:

        github tools: (44)
        - create_issue: Open a new issue in a GitHub repository.
        - merge_pull_request: Merge an open pull request.
        ...

    Ordering is deterministic (groups and tools sorted by name) so the
    rendered block is byte-stable across assemblies of the same catalog —
    this keeps the request prefix cacheable across turns.

    Token-budget fallbacks (cheap chars/4 estimate, same rule as the
    activation gate):
      1. full listing (names + short descriptions)
      2. names-only listing, still grouped
      3. server-level summary — one line per MCP server / plugin toolset
         (name + tool count), so the model always knows WHICH domains are
         reachable through the bridge even when per-tool names don't fit
      4. ``None`` — only when the summary itself exceeds the budget
    """
    text, _form = build_catalog_listing_with_form(deferrable, max_tokens=max_tokens)
    return text
# ---- END PLUGIN-COMPAT ----