hermes-agent/tools/tool_search_validation.py

"""Local argument validation for ``tool_call`` against a deferred tool's schema."""

from __future__ import annotations

import copy
import json
import logging
import re
from typing import Any, Dict, Optional

from tools.registry import tool_error

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

_SCHEMA_LITERAL_KEYS = frozenset({"const", "default", "enum", "example", "examples"})


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."""
    if isinstance(node, list):
        return [_schema_for_local_validation(item) for item in node]
    if not isinstance(node, dict):
        return node

    normalized = {}
    for key, value in node.items():
        if key == "nullable":
            continue
        # Literal keywords hold instance data, not schemas: copy byte-for-byte.
        normalized[key] = (
            copy.deepcopy(value)
            if key in _SCHEMA_LITERAL_KEYS
            else _schema_for_local_validation(value)
        )
    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
    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.
    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)."""
    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()
        if key not in _SCHEMA_LITERAL_KEYS
    )


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):
            path += f".{part}"
        else:
            path += f"[{json.dumps(part, ensure_ascii=False)}]"
    return path


def _validation_error(message: str, *, path: str, constraint: str, parameters: Any) -> str:
    return tool_error(
        message,
        path=path,
        constraint=constraint,
        parameters=parameters,
        hint="Retry tool_call with 'arguments' matching the parameters schema above.",
    )


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.
    """
    try:
        from tools.registry import registry as _registry
        schema = _registry.get_schema(name)
        if not isinstance(schema, dict):
            return None
        fn = schema.get("function") if schema.get("type") == "function" else schema
        params = fn.get("parameters") if isinstance(fn, dict) else None
        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,
                )

        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).
        candidate_args = dict(args)
        try:
            from model_tools import coerce_tool_args
            candidate_args = coerce_tool_args(name, candidate_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)
        )
        if validation_error is None:
            return None

        path = _validation_path(validation_error)
        constraint = str(getattr(validation_error, "validator", None) or "schema")
        detail = re.sub(r"\s+", " ", str(validation_error.message)).strip()
        if len(detail) > 600:
            detail = detail[:597] + "..."
        return _validation_error(
            f"tool_call to '{name}' failed argument validation at {path} "
            f"({constraint}): {detail}. The tool was NOT invoked.",
            path=path, constraint=constraint, parameters=params,
        )
    except Exception:  # pragma: no cover — never block dispatch on validator bugs
        logger.debug("validate_deferred_call_args failed for %s", name, exc_info=True)
        return None