hermes-agent/tools/mcp_tool_agent.py

"""Live-agent tool-list maintenance after MCP (re)discovery: refreshing an
AIAgent's tools/tool names, preserving the cached tools[] prefix across rebuilds,
and re-injecting post-build tools."""

import logging
import json
import threading
from typing import Optional
from tools.mcp_tool_common import _core

logger = logging.getLogger("tools.mcp_tool")

# Serializes in-place swaps of ``agent.tools`` / ``agent.valid_tool_names`` by
# the reload RPC, gateway reload and late-binding refresh thread; the run loop
# reads them during tool iteration and must never see a half-updated pair.
_agent_tools_lock = threading.Lock()


def _def_name(tool_def: dict) -> str:
    return (tool_def.get("function") or {}).get("name", "")


def _agent_tool_defs(agent) -> list:
    return list(getattr(agent, "tools", None) or [])


def agent_tool_names(agent) -> list:
    """Names of ``agent.tools`` in wire order (unnamed entries skipped)."""
    return [name for name in map(_def_name, _agent_tool_defs(agent)) if name]


def _resolve_refresh_toolsets(agent, enabled_override, disabled_override):
    """Explicit reloads pass freshly-resolved toolsets (so a server just ENABLED in config is
    picked up) and the agent's selection is updated to match; automatic paths pass nothing
    and reuse the build-time selection."""
    enabled = getattr(agent, "enabled_toolsets", None)
    disabled = getattr(agent, "disabled_toolsets", None)
    if enabled_override is not None or disabled_override is not None:
        enabled = enabled_override if enabled_override is not None else enabled
        disabled = disabled_override if disabled_override is not None else disabled
        agent.enabled_toolsets, agent.disabled_toolsets = enabled, disabled
    return enabled, disabled


def _tool_defs_content_changed(agent, new_defs: list) -> bool:
    """Byte-level diff of the serialized tool arrays (dynamic schemas change CONTENT under
    stable names); False if either side fails to serialize."""
    try:
        dump = lambda defs: json.dumps(defs, sort_keys=True, separators=(",", ":"), default=str)  # noqa: E731
        return dump(_agent_tool_defs(agent)) != dump(new_defs)
    except Exception:  # noqa: BLE001
        return False


def _drop_side_agent_tools(agent, new_defs: list, new_names: set) -> tuple:
    from tools.connectors.turn import side_agent_tool_drops
    drops = side_agent_tool_drops(agent)
    if not drops:
        return new_defs, new_names
    return [entry for entry in new_defs if _def_name(entry) not in drops], new_names - drops


def _publish_tool_snapshot(
    agent, new_defs: list, new_names: set, *, snapshot_generation: int,
    staged_engine_names: set, content_aware: bool, prefix_registered: Optional[set]) -> Optional[set]:
    """Single atomic read-diff-publish under ``_agent_tools_lock`` so ``added`` matches what
    was published and a stale (older-generation) rebuild can't overwrite a newer one. Returns
    the added names, or None when nothing was published (unchanged, or a newer snapshot won)."""
    with _agent_tools_lock:
        # Tolerate an agent that never set the generation (or a non-int mock).
        published_gen = getattr(agent, "_tool_snapshot_generation", -1)
        published_gen = published_gen if isinstance(published_gen, int) else -1
        if snapshot_generation < published_gen:
            return None  # a newer snapshot already won
        current_defs = _agent_tool_defs(agent)
        current = {_def_name(t) for t in current_defs}
        if prefix_registered is not None:
            new_defs, new_names = _merge_preserving_prefix(current_defs, new_defs, prefix_registered)
        new_defs, new_names = _drop_side_agent_tools(agent, new_defs, new_names)
        # Record the generation even when unchanged so an in-flight older caller can't clobber.
        agent._tool_snapshot_generation = max(published_gen, snapshot_generation)
        # Same NAME set: no change for MCP-reload callers. Content-aware callers
        # (compaction boundary) also diff serialized bytes.
        if new_names == current and not (content_aware and _tool_defs_content_changed(agent, new_defs)):
            return None
        agent.tools = new_defs
        agent.valid_tool_names = new_names
        # Publish context-engine routing names atomically with the snapshot.
        engine_names = getattr(agent, "_context_engine_tool_names", None)
        if isinstance(engine_names, set):
            engine_names.clear()
            engine_names.update(staged_engine_names)
        return new_names - current


def refresh_agent_mcp_tools(
    agent, *, enabled_override=None, disabled_override=None, quiet_mode: bool = True,
    content_aware: bool = False, preserve_prefix: bool = False) -> set:
    """Re-derive an already-built agent's tool snapshot from the live registry; returns the
    newly-added tool names (empty when unchanged). The agent snapshots ``agent.tools`` at build
    time, so servers that connect later (slow OAuth, ``/reload-mcp``) are invisible until
    rebuilt. Shared by the TUI RPC, gateway reload, late-binding thread and between-turns
    refresh: respects the toolset filter, diffs by tool NAME (a count compare misses an
    equal-size swap), re-injects the memory-provider / context-engine tools ``agent_init``
    appends after ``get_tool_definitions`` plus guarded session capabilities, and publishes
    ``(tools, valid_tool_names)`` together.

    ``preserve_prefix``: for rebuilds inside a live conversation the tool array is a cached
    request prefix and any moved byte re-prefills the whole history — existing tools keep their
    slot (schemas still refresh), a still-registered tool whose ``check_fn`` merely flapped is
    carried forward (``check_fn`` gates exposure, never invocation), a deregistered tool is
    dropped, new tools append at the tail. The caller owns the prompt-cache contract."""
    from model_tools import get_tool_definitions
    from tools.registry import registry
    enabled, disabled = _resolve_refresh_toolsets(agent, enabled_override, disabled_override)
    # Generation captured BEFORE the slow get_tool_definitions call (a slower caller holding an
    # OLDER set must not clobber a newer one); definitions computed OUTSIDE the lock.
    snapshot_generation = registry._generation
    new_defs = list(get_tool_definitions(enabled_toolsets=enabled, disabled_toolsets=disabled, quiet_mode=quiet_mode) or [])
    new_names = {_def_name(t) for t in new_defs}
    # Post-build families re-appended on LOCALS only; live attributes untouched until publish.
    staged_engine_names = _reinject_post_build_tools(agent, new_defs, new_names)
    _reinject_authorized_dynamic_tools(agent, new_defs, new_names)
    # Registry membership is read OUTSIDE ``_agent_tools_lock``: taking ``registry._lock``
    # under the tools lock would be the first nesting of the two.
    prefix_registered: Optional[set] = None
    if preserve_prefix:
        try:
            prefix_registered = {entry.name for entry in registry.get_all_entries()}
        except Exception:  # noqa: BLE001
            pass  # fail open to the plain rebuild
    added = _publish_tool_snapshot(
        agent, new_defs, new_names, snapshot_generation=snapshot_generation,
        staged_engine_names=staged_engine_names, content_aware=content_aware, prefix_registered=prefix_registered)
    if added is None:
        return set()
    persist_agent_tool_names(agent)  # re-pin so a rebuild after agent-cache eviction restores this order
    return added


def reprobe_tool_availability() -> None:
    """Explicit ``/reload-mcp`` hatch out of the tools[] freeze: drop the ``check_fn`` verdict
    cache AND the ``get_tool_definitions`` memo (keyed on registry generation, so it would
    otherwise replay the stale verdicts)."""
    from model_tools import _clear_tool_defs_cache
    from tools.registry import invalidate_check_fn_cache
    invalidate_check_fn_cache()
    _clear_tool_defs_cache()


def tool_pin_version() -> str:
    """The code identity a tools[] pin was built by (checkout/build sha, else the release version).
    Cached per process: an updated checkout only reaches a process through a restart."""
    from hermes_cli import __version__
    from hermes_cli.build_info import get_code_identity
    identity = get_code_identity()
    return identity.get("sha") or identity.get("version") or __version__


def persist_agent_tool_names(agent) -> None:
    """Best-effort: write ``agent.tools`` to the session row (freeze pin). The full definitions,
    keyed by the code that built them: another process or surface derives different bytes."""
    db = getattr(agent, "_session_db", None)
    session_id = getattr(agent, "session_id", None)
    if not db or not session_id:
        return
    try:
        db.update_session_tool_names(session_id, {"version": tool_pin_version(), "tools": _agent_tool_defs(agent)})
    except Exception:  # noqa: BLE001
        logger.debug("tool_names persist skipped", exc_info=True)


def _config_permitted_names(agent) -> set:
    """Tool names this agent's toolset selection allows before ``check_fn``: all a pin may carry
    forward. A client-surface toolset counts as allowed (only its client can add it, so its absence
    here is no config choice); ``disabled_toolsets`` and role reservations still strip it."""
    from model_tools import _select_tool_names
    from toolsets import CLIENT_SURFACE_TOOLSETS
    enabled = getattr(agent, "enabled_toolsets", None)
    if enabled is not None:
        enabled = [*enabled, *CLIENT_SURFACE_TOOLSETS]
    return _select_tool_names(enabled, getattr(agent, "disabled_toolsets", None), True)


def _drop_gated_carried_tools(merged: list, carried: set) -> list:
    """A carried tool also passes the session-level schema gates the fresh build applied
    (``browser_exec`` needs ``terminal`` in the same array), judged on the merged array."""
    from model_tools import _DYNAMIC_SCHEMA_REWRITERS
    available = {_def_name(t) for t in merged}
    return [t for t in merged if _def_name(t) not in carried or _def_name(t) not in _DYNAMIC_SCHEMA_REWRITERS
            or _DYNAMIC_SCHEMA_REWRITERS[_def_name(t)](t, available) is not None]


def restore_agent_tool_prefix(agent, saved) -> bool:
    """Fold a freshly built agent's ``tools`` onto the session's pin; True if changed.
    A fresh AIAgent (gateway cache eviction, ``--resume`` in a new process, a surface hop) has no
    predecessor to preserve, so the pin stands in. Pinned by the SAME code, a tool still available
    here keeps its pinned BYTES, whatever this process derives for it (tool_search's per-surface
    catalog, per-surface dynamic parameters, the ``-q`` footprint): tools[] heads every request.
    Pinned by other code (``hermes update``, a legacy name list) a tool's contract may have moved,
    so each takes its current definition. A pinned tool this process did not build is carried
    only while its toolset config allows it here; deregistered tools drop, new tools append."""
    pinned, version = (saved.get("tools") or [], saved.get("version")) if isinstance(saved, dict) else (saved, None)
    if not pinned:
        return False
    from tools.registry import registry
    fresh_defs = _agent_tool_defs(agent)
    fresh = {_def_name(t): t for t in fresh_defs}
    registered_names = {entry.name for entry in registry.get_all_entries()}
    same_code = version is not None and version == tool_pin_version()

    def _pinned_def(item):
        name = item if isinstance(item, str) else _def_name(item)
        if isinstance(item, dict) and same_code:
            return item
        if name in fresh:
            return fresh[name]
        entry = registry.get_entry(name)
        return None if entry is None else {"type": "function", "function": {**entry.schema, "name": entry.name}}

    pinned_defs = [d for d in map(_pinned_def, pinned) if d]
    carried = {_def_name(d) for d in pinned_defs if _def_name(d) not in fresh and _def_name(d) in registered_names}
    carried &= _config_permitted_names(agent) if carried else set()
    merged = [d for d in pinned_defs if _def_name(d) in fresh or _def_name(d) in carried]
    pinned_names = {_def_name(d) for d in merged}
    merged.extend(t for t in fresh_defs if _def_name(t) not in pinned_names)
    merged = _drop_gated_carried_tools(merged, carried)
    merged_names = {_def_name(t) for t in merged}
    _reinject_authorized_dynamic_tools(agent, merged, merged_names)
    merged, merged_names = _drop_side_agent_tools(agent, merged, merged_names)
    changed = merged != fresh_defs
    if changed:
        with _agent_tools_lock:
            agent.tools = merged
            agent.valid_tool_names = merged_names
    if not same_code or merged != list(pinned):
        persist_agent_tool_names(agent)
    return changed


def _merge_preserving_prefix(current_defs: list, new_defs: list, registered_names: set) -> tuple[list, set]:
    """Fold a fresh tool snapshot into a live one without moving existing bytes. Ordered by
    ``current_defs`` (the cached request prefix): a name in both keeps its slot but takes the
    fresh schema; a name only in the live list is kept if still registered (``check_fn``
    flapped), else dropped; a name only in the fresh list is appended at the tail.

    The bridge tools keep their BUILT entry, not the fresh one: ``tool_search``'s description
    is derived from the session (deferred count, listing, whether ``manage_connections`` was
    present), so a late MCP server or a ``check_fn`` flap would rewrite it every turn. Search
    reads the live catalog at dispatch, so the stale count costs nothing."""
    from tools.tool_search_catalog import BRIDGE_TOOL_NAMES
    fresh = {_def_name(entry): entry for entry in new_defs if _def_name(entry)}
    merged = []
    for entry in current_defs:
        name = _def_name(entry)
        replacement = fresh.pop(name, None)
        if name in BRIDGE_TOOL_NAMES:
            merged.append(entry)
        elif replacement is not None:
            merged.append(replacement)
        elif name and name in registered_names:
            merged.append(entry)
    merged.extend(fresh.values())
    return merged, {_def_name(t) for t in merged}


def _reinject_authorized_dynamic_tools(agent, tools_list: list, name_set: set) -> None:
    """``message_agent`` is injected by an auth gate, never registered, so a registry-derived
    rebuild drops it. Scrub any stale copy from the STAGED pair and re-add it only when the live
    gate re-authorizes, so the publisher exposes a coherent ``(tools, valid_tool_names)``."""
    from tools.bot_mode_dm import MESSAGE_AGENT_TOOL_NAME, message_agent_authorized, message_agent_tool_schema

    tools_list[:] = [entry for entry in tools_list if _def_name(entry) != MESSAGE_AGENT_TOOL_NAME]
    name_set.discard(MESSAGE_AGENT_TOOL_NAME)
    if message_agent_authorized(agent):
        tools_list.append(message_agent_tool_schema())
        name_set.add(MESSAGE_AGENT_TOOL_NAME)


def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
    """Append memory-provider and context-engine tools onto the caller's staged ``tools_list``
    / ``name_set`` (never the live agent attributes), mirroring ``agent_init``'s post-build
    injection. Idempotent and fail-soft. Returns the context-engine routing names THIS rebuild
    appended: a name already owned by a registry/plugin tool is not claimed, matching agent_init."""
    def _add(schema) -> bool:
        name = schema.get("name", "") if isinstance(schema, dict) else ""
        if not name or name in name_set:
            return False
        tools_list.append({"type": "function", "function": schema})
        name_set.add(name)
        return True

    def _schema_getter(attr: str, method: str):
        getter = getattr(getattr(agent, attr, None) or None, method, None)
        return getter if callable(getter) else None

    enabled = getattr(agent, "enabled_toolsets", None)
    try:
        get_mem_schemas = _schema_getter("_memory_manager", "get_all_tool_schemas")
        if get_mem_schemas is not None:
            from agent.memory_manager import memory_provider_tools_enabled  # same gate inject_memory_provider_tools uses
            if memory_provider_tools_enabled(
                    enabled, getattr(agent, "disabled_toolsets", None), memory_tool_present="memory" in name_set):
                for schema in get_mem_schemas():
                    _add(schema)
    except Exception:
        logger.debug("Memory-provider tool re-injection skipped", exc_info=True)

    # The `context_engine` toolset is intentionally empty, so lcm_* tools exist only via this
    # append. Honor the enabled_toolsets gate agent_init uses, or a restricted-toolset platform
    # would re-leak tools the build excluded.
    # See #5544.
    staged_engine_names: set = set()
    try:
        get_schemas = _schema_getter("context_compressor", "get_tool_schemas")
        if (enabled is None or "context_engine" in enabled) and get_schemas is not None:
            # Claim the routing name only when WE appended the schema.
            staged_engine_names.update(s["name"] for s in get_schemas() if _add(s))
    except Exception:
        logger.debug("Context-engine tool re-injection skipped", exc_info=True)
    return staged_engine_names