diff --git a/tools/url_safety.py b/tools/url_safety.py index bf62f8be5b..88dbdc8b83 100644 --- a/tools/url_safety.py +++ b/tools/url_safety.py @@ -1,18 +1,10 @@ """URL safety checks — blocks requests to private/internal network addresses (SSRF). -``security.allow_private_urls: true`` (config.yaml) disables private-IP blocking -for environments whose DNS resolves public names to private/benchmark ranges. -Even then, cloud metadata hostnames/IPs are **always** blocked. - -Limitations: - - DNS rebinding (TOCTOU): an attacker DNS server with TTL=0 can answer a public - IP for the check and a private one for the connect. Hermes-owned direct httpx - paths should use ``create_ssrf_safe_client()`` / ``create_ssrf_safe_async_client()`` - so the policy is re-applied at TCP connect and the socket dials the validated - IP while preserving Host/SNI semantics. - - Redirect bypass is mitigated by httpx response hooks re-validating each - redirect target (see ``redirect_target_from_response``). Web tools go through - third-party SDKs (Firecrawl/Tavily) whose redirect handling is server-side. +``security.allow_private_urls: true`` disables private-IP blocking (DNS that resolves public +names to private ranges); cloud metadata hostnames/IPs are **always** blocked. DNS rebinding +(TOCTOU) is closed for Hermes-owned httpx paths by ``create_ssrf_safe_[async_]client()``, which +re-apply the policy at TCP connect and dial the validated IP while preserving Host/SNI. Redirect +bypass is mitigated by response hooks re-validating each target (``redirect_target_from_response``). """ import ipaddress @@ -21,6 +13,7 @@ import os import socket import asyncio import re +from contextlib import contextmanager from typing import Any, Optional from urllib.parse import parse_qsl, quote, unquote, urljoin, urlparse, urlsplit, urlunsplit @@ -29,15 +22,10 @@ from utils import is_truthy_value logger = logging.getLogger(__name__) - # Proxy env vars: when set, the runtime should delegate DNS to the proxy. -_PROXY_ENV_VARS = ( - "HTTPS_PROXY", "https_proxy", - "HTTP_PROXY", "http_proxy", - "ALL_PROXY", "all_proxy", -) - +_PROXY_ENV_VARS = ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy", "ALL_PROXY", "all_proxy") _HTTP_SCHEMES = frozenset({"http", "https"}) +_IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address def _proxy_is_configured() -> bool: @@ -45,16 +33,11 @@ def _proxy_is_configured() -> bool: def normalize_url_for_request(url: str) -> str: - """Return an ASCII-safe HTTP URL for Hermes-owned URL tools (IRI -> URI). - - Browsers expect URIs but users/models often supply IRIs (``https://wttr.in/Köln``). - Preserves URL syntax and existing percent escapes while IDNA-encoding the host - and percent-encoding non-ASCII path/query/fragment text. URL tool inputs only — - arbitrary shell commands must not be rewritten. - """ + """ASCII-safe HTTP URL for Hermes-owned URL tools (IRI -> URI, e.g. ``https://wttr.in/Köln``). + Preserves URL syntax and existing percent escapes while IDNA-encoding the host and + percent-encoding non-ASCII path/query/fragment text. URL tool inputs only — never shell commands.""" if not isinstance(url, str): return url - raw = url.strip() if not raw: return raw @@ -62,17 +45,13 @@ def normalize_url_for_request(url: str) -> str: # Repair model-emitted whitespace between scheme separator and authority # (``https:// docs.example``); that position is never meaningful in HTTP URLs. raw = re.sub(r"^([A-Za-z][A-Za-z0-9+.-]*://)\s+", r"\1", raw) - try: parsed = urlsplit(raw) except ValueError: return raw - if parsed.scheme.lower() not in _HTTP_SCHEMES: return raw - - netloc = parsed.netloc - hostname = parsed.hostname + netloc, hostname = parsed.netloc, parsed.hostname if hostname: try: ascii_host = hostname.encode("idna").decode("ascii") @@ -80,48 +59,25 @@ def normalize_url_for_request(url: str) -> str: ascii_host = hostname if ascii_host != hostname: netloc = netloc.replace(hostname, ascii_host, 1) - - path = quote(parsed.path, safe="/%:@!$&'()*+,;=") - query = quote(parsed.query, safe="/%:@!$&'()*+,;=?") - fragment = quote(parsed.fragment, safe="/%:@!$&'()*+,;=?") - - return urlunsplit((parsed.scheme, netloc, path, query, fragment)) + safe = "/%:@!$&'()*+,;=" + return urlunsplit((parsed.scheme, netloc, quote(parsed.path, safe=safe), + quote(parsed.query, safe=safe + "?"), quote(parsed.fragment, safe=safe + "?"))) # Unambiguously credential-bearing query param names. Deliberately narrow: bare # English words that double as page facets (``code``, ``key``, ``auth``, # ``session``, ``sig``) are EXCLUDED so ordinary browsing is not blocked. _SENSITIVE_QUERY_PARAM_NAMES = frozenset({ - "access_token", - "api_key", - "apikey", - "auth_token", - "authorization", - "awsaccesskeyid", - "client_secret", - "credential", - "credentials", - "jwt", - "password", - "passwd", - "secret", - "session_id", - "signature", - "token", - "x_amz_security_token", - "x_amz_signature", - "x-amz-security-token", - "x-amz-signature", -}) + "access_token", "api_key", "apikey", "auth_token", "authorization", "awsaccesskeyid", + "client_secret", "credential", "credentials", "jwt", "password", "passwd", "secret", + "session_id", "signature", "token", "x_amz_security_token", "x_amz_signature", + "x-amz-security-token", "x-amz-signature"}) def sensitive_query_param_name(url: str) -> Optional[str]: - """Return the first credential-named query parameter in ``url`` (with a value), if any. - - Checked before handing URLs to third-party fetch/browser backends: prefix-based - token redaction catches known vendor key shapes; this catches opaque magic links, - OAuth codes, signed-URL signatures and custom ``?token=...`` values. - """ + """First credential-named query parameter in ``url`` (with a value), if any. Checked before + handing URLs to third-party fetch/browser backends: catches opaque magic links, OAuth codes, + signed-URL signatures and custom ``?token=...`` values that prefix-based redaction misses.""" if not isinstance(url, str) or "?" not in url: return None try: @@ -130,43 +86,32 @@ def sensitive_query_param_name(url: str) -> Optional[str]: return None if parsed.scheme.lower() not in _HTTP_SCHEMES or not parsed.query: return None - for key, value in parse_qsl(parsed.query, keep_blank_values=True): - if value and unquote(key).lower() in _SENSITIVE_QUERY_PARAM_NAMES: - return key - return None + return next((key for key, value in parse_qsl(parsed.query, keep_blank_values=True) + if value and unquote(key).lower() in _SENSITIVE_QUERY_PARAM_NAMES), None) # Cloud metadata hostnames — always blocked regardless of DNS or config toggle. -_BLOCKED_HOSTNAMES = frozenset({ - "metadata.google.internal", - "metadata.goog", -}) +_BLOCKED_HOSTNAMES = frozenset({"metadata.google.internal", "metadata.goog"}) -# Cloud metadata / credential endpoints (the #1 SSRF target) and the link-local -# range they live in — always blocked. IPv4-mapped IPv6 forms are listed because -# resolvers may return ``::ffff:x.x.x.x`` and ipaddress treats those as distinct. -_ALWAYS_BLOCKED_IPS = frozenset({ - ipaddress.ip_address("169.254.169.254"), # AWS/GCP/Azure/DO/Oracle metadata - ipaddress.ip_address("169.254.170.2"), # AWS ECS task metadata (task IAM creds) - ipaddress.ip_address("169.254.169.253"), # Azure IMDS wire server - ipaddress.ip_address("fd00:ec2::254"), # AWS metadata (IPv6) - ipaddress.ip_address("100.100.100.200"), # Alibaba Cloud metadata - ipaddress.ip_address("::ffff:169.254.169.254"), - ipaddress.ip_address("::ffff:169.254.170.2"), - ipaddress.ip_address("::ffff:169.254.169.253"), - ipaddress.ip_address("::ffff:100.100.100.200"), -}) -_ALWAYS_BLOCKED_NETWORKS = ( - ipaddress.ip_network("169.254.0.0/16"), # Entire link-local range (no legit agent target) - ipaddress.ip_network("::ffff:169.254.0.0/112"), # IPv4-mapped link-local range +# Cloud metadata / credential endpoints (the #1 SSRF target) — always blocked, also in +# IPv4-mapped IPv6 form (resolvers may return ``::ffff:x.x.x.x``; ipaddress treats those as distinct). +_METADATA_V4 = ( + "169.254.169.254", # AWS/GCP/Azure/DO/Oracle metadata + "169.254.170.2", # AWS ECS task metadata (task IAM creds) + "169.254.169.253", # Azure IMDS wire server + "100.100.100.200", # Alibaba Cloud metadata ) +_ALWAYS_BLOCKED_IPS = frozenset( + {ipaddress.ip_address(ip) for ip in _METADATA_V4} + | {ipaddress.ip_address("::ffff:" + ip) for ip in _METADATA_V4} + | {ipaddress.ip_address("fd00:ec2::254")} # AWS metadata (IPv6) +) +# Entire link-local range (no legit agent target), plus its IPv4-mapped form. +_ALWAYS_BLOCKED_NETWORKS = tuple(ipaddress.ip_network(n) for n in ("169.254.0.0/16", "::ffff:169.254.0.0/112")) # Exact HTTPS hostnames allowed to resolve to private/benchmark-space IPs # (QQ media legitimately resolves to 198.18.0.0/15 behind local proxy infra). -_TRUSTED_PRIVATE_IP_HOSTS = frozenset({ - "multimedia.nt.qq.com.cn", -}) - +_TRUSTED_PRIVATE_IP_HOSTS = frozenset({"multimedia.nt.qq.com.cn"}) _MAX_SSRF_CONNECT_IPS = 8 # 100.64.0.0/10 (CGNAT, RFC 6598) is neither is_private nor is_global in @@ -174,30 +119,19 @@ _MAX_SSRF_CONNECT_IPS = 8 _CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10") # Global toggle cache (process lifetime; see _global_allow_private_urls). -_allow_private_resolved = False -_cached_allow_private: bool = False +_allow_private_resolved, _cached_allow_private = False, False def _global_allow_private_urls() -> bool: - """Return True when the user has opted out of private-IP blocking. - - Priority: ``HERMES_ALLOW_PRIVATE_URLS`` env, ``security.allow_private_urls``, - legacy ``browser.allow_private_urls``. A multiplex gateway serves several - independently configured profiles in one process, so profile-scoped turns - (``get_hermes_home_override()`` set) bypass the process-global cache — - otherwise the first profile's opt-out would disable blocking for every later one. - ``read_raw_config()`` already provides path/mtime caching for that path. - """ + """True when the user has opted out of private-IP blocking. Priority: ``HERMES_ALLOW_PRIVATE_URLS`` + env, ``security.allow_private_urls``, legacy ``browser.allow_private_urls``. Profile-scoped turns + (``get_hermes_home_override()`` set) bypass the process-global cache — a multiplex gateway serves + several profiles in one process; the first profile's opt-out must not disable blocking for later ones.""" global _allow_private_resolved, _cached_allow_private - if get_hermes_home_override() is not None: return _resolve_allow_private_urls() - - if _allow_private_resolved: - return _cached_allow_private - - _allow_private_resolved = True - _cached_allow_private = _resolve_allow_private_urls() + if not _allow_private_resolved: + _allow_private_resolved, _cached_allow_private = True, _resolve_allow_private_urls() return _cached_allow_private @@ -208,35 +142,30 @@ def _resolve_allow_private_urls() -> bool: return True if env_val in {"false", "0", "no"}: return False # explicit false does not fall through to config - try: from hermes_cli.config import read_raw_config cfg = read_raw_config() for section in ("security", "browser"): # preferred, then legacy block = cfg.get(section, {}) - if isinstance(block, dict) and is_truthy_value( - block.get("allow_private_urls"), default=False - ): + if isinstance(block, dict) and is_truthy_value(block.get("allow_private_urls"), default=False): return True except Exception: pass # config unavailable (tests, early import) — keep default - return False def _reset_allow_private_cache() -> None: """Reset the cached toggle — only for tests.""" global _allow_private_resolved, _cached_allow_private - _allow_private_resolved = False - _cached_allow_private = False + _allow_private_resolved = _cached_allow_private = False def _normalize_hostname(host: Optional[str]) -> str: return (host or "").strip().lower().rstrip(".") -def _parse_ip(hostname: str) -> Optional[ipaddress.IPv4Address | ipaddress.IPv6Address]: - """Return the IP object for a literal-IP hostname, else None.""" +def _parse_ip(hostname: str) -> Optional[_IPAddress]: + """IP object for a literal-IP hostname, else None.""" try: return ipaddress.ip_address(hostname) except ValueError: @@ -244,22 +173,23 @@ def _parse_ip(hostname: str) -> Optional[ipaddress.IPv4Address | ipaddress.IPv6A def _iter_resolved_ips(addr_info: Any): - """Yield ``(raw, ip_str, ip)`` per getaddrinfo answer. - - ``ip_str`` has any IPv6 scope ID (``%eth0``) stripped; ``ip`` is None when the - answer is still unparseable — each caller decides skip / fail-closed / raise. - """ + """Yield ``(raw, ip_str, ip)`` per getaddrinfo answer. ``ip_str`` has any IPv6 scope ID + (``%eth0``) stripped; ``ip`` is None when still unparseable — each caller decides skip/fail/raise.""" for _family, _, _, _, sockaddr in addr_info: raw = sockaddr[0] - ip_str = raw.split("%")[0] if "%" in raw else raw + ip_str = raw.split("%")[0] yield raw, ip_str, _parse_ip(ip_str) -def _is_always_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: +def _getaddrinfo(hostname: str, port: Optional[int] = None): + return socket.getaddrinfo(hostname, port, socket.AF_UNSPEC, socket.SOCK_STREAM) + + +def _is_always_blocked_ip(ip: _IPAddress) -> bool: return ip in _ALWAYS_BLOCKED_IPS or any(ip in net for net in _ALWAYS_BLOCKED_NETWORKS) -def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: +def _is_blocked_ip(ip: _IPAddress) -> bool: """Return True if the IP should be blocked for SSRF protection.""" # IPv4-mapped IPv6 (``::ffff:x.x.x.x``) is classified by its embedded IPv4. if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: @@ -269,36 +199,28 @@ def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: def is_always_blocked_url(url: str) -> bool: - """Return True when the URL targets the always-blocked floor (cloud metadata). - - Narrower than ``is_safe_url``: only the sentinel hostnames/IPs, regardless of - backend, routing, or ``allow_private_urls``. For callers that deliberately bypass - the full check (e.g. hybrid cloud browser routing private URLs to a local sidecar) - but must still enforce the non-negotiable floor. Returns False for ordinary - private/loopback URLs, DNS failures on non-sentinel hosts, and parse errors - (the caller's ordinary fail-closed path handles those). - """ + """True when the URL targets the always-blocked floor (cloud metadata) — only the sentinel + hostnames/IPs, regardless of backend, routing, or ``allow_private_urls``. For callers that + deliberately bypass the full check (e.g. hybrid cloud browser routing private URLs to a local + sidecar) but must still enforce the floor. False for ordinary private/loopback URLs, DNS + failures and parse errors (the caller's ordinary fail-closed path handles those).""" try: hostname = _normalize_hostname(urlparse(url).hostname) if not hostname: return False - if hostname in _BLOCKED_HOSTNAMES: logger.warning("Blocked request to internal hostname (always-blocked floor): %s", hostname) return True - ip = _parse_ip(hostname) if ip is not None: if _is_always_blocked_ip(ip): logger.warning("Blocked request to cloud metadata address (always-blocked floor): %s", hostname) return True return False - try: - addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) + addr_info = _getaddrinfo(hostname) except socket.gaierror: return False # DNS failure is not part of the floor; caller's path handles it - for raw, ip_str, resolved in _iter_resolved_ips(addr_info): if resolved is None: logger.warning("Unparseable IP address %r for hostname %s — skipping address", raw, hostname) @@ -308,9 +230,7 @@ def is_always_blocked_url(url: str) -> bool: "Blocked request to cloud metadata address (always-blocked floor): %s -> %s", hostname, ip_str ) return True - return False - except Exception as exc: # Parse/unexpected errors are not "always blocked"; caller decides fail-open/closed. logger.debug("is_always_blocked_url error for %s: %s", url, exc) @@ -322,14 +242,9 @@ def _allows_private_ip_resolution(hostname: str, scheme: str) -> bool: return scheme == "https" and hostname in _TRUSTED_PRIVATE_IP_HOSTS -def _resolved_ip_block_reason( - ip: ipaddress.IPv4Address | ipaddress.IPv6Address, allow_private: bool -) -> Optional[str]: - """Why a resolved answer must be rejected, or None if it may be dialed. - - The metadata floor is checked first and ignores ``allow_private``; ordinary - private/internal classes are only blocked when ``allow_private`` is False. - """ +def _resolved_ip_block_reason(ip: _IPAddress, allow_private: bool) -> Optional[str]: + """Why a resolved answer must be rejected, or None if it may be dialed. The metadata floor + ignores ``allow_private``; ordinary private/internal classes are blocked only when it is False.""" if _is_always_blocked_ip(ip): return "cloud metadata address" if not allow_private and _is_blocked_ip(ip): @@ -338,12 +253,9 @@ def _resolved_ip_block_reason( def is_safe_url(url: str) -> bool: - """Return True if the URL target is not a private/internal address. - - Resolves the hostname and checks every answer. Fails closed on DNS errors and - unexpected exceptions. ``allow_private_urls`` skips private-IP blocking, but - cloud metadata endpoints remain blocked regardless. - """ + """True if the URL target is not a private/internal address. Resolves the hostname and checks + every answer; fails closed on DNS errors and unexpected exceptions. ``allow_private_urls`` + skips private-IP blocking, but cloud metadata endpoints remain blocked regardless.""" try: parsed = urlparse(url) hostname = _normalize_hostname(parsed.hostname) @@ -358,13 +270,11 @@ def is_safe_url(url: str) -> bool: if hostname in _BLOCKED_HOSTNAMES: logger.warning("Blocked request to internal hostname: %s", hostname) return False - allow_all_private = _global_allow_private_urls() allow_private_ip = _allows_private_ip_resolution(hostname, scheme) allow_private = allow_all_private or allow_private_ip - try: - addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) + addr_info = _getaddrinfo(hostname) except socket.gaierror: # Sandbox/proxy environments may block direct DNS; when a proxy is # configured, delegate resolution to it (metadata hostnames were already @@ -373,29 +283,23 @@ def is_safe_url(url: str) -> bool: if _parse_ip(hostname) is None and _proxy_is_configured(): logger.debug( "DNS resolution failed for %s — proxy configured, allowing through for proxy-side resolution", - hostname, - ) + hostname) return True logger.warning("Blocked request — DNS resolution failed for: %s", hostname) return False - for raw, ip_str, ip in _iter_resolved_ips(addr_info): if ip is None: logger.warning("Blocked request — unparseable IP address %r for hostname %s", raw, hostname) return False - reason = _resolved_ip_block_reason(ip, allow_private) if reason is not None: logger.warning("Blocked request to %s: %s -> %s", reason, hostname, ip_str) return False - if allow_all_private: logger.debug("Allowing private/internal resolution (security.allow_private_urls=true): %s", hostname) elif allow_private_ip: logger.debug("Allowing trusted hostname despite private/internal resolution: %s", hostname) - return True - except Exception as exc: # Fail closed: parsing edge cases must not become SSRF bypass vectors. logger.warning("Blocked request — URL safety check error for %s: %s", url, exc) @@ -403,7 +307,7 @@ def is_safe_url(url: str) -> bool: async def async_is_safe_url(url: str) -> bool: - """Same rules as :func:`is_safe_url`, with the blocking DNS work off the event loop.""" + """:func:`is_safe_url` with the blocking DNS work off the event loop.""" return await asyncio.to_thread(is_safe_url, url) @@ -411,67 +315,53 @@ class SSRFConnectionBlocked(ValueError): """Raised when connect-time DNS resolution violates the URL safety policy.""" -def _safe_connect_scheme(host: str, port: int, schemes_by_origin: dict[tuple[str, int], str]) -> str: - return schemes_by_origin.get((host, port)) or ("https" if port == 443 else "http") - - def _resolved_http_connect_ips(host: str, port: int, scheme: str) -> list[str]: - """Resolve and validate *host* at TCP-connect time; return dialable IP strings. - - Closes the DNS-rebinding gap between pre-flight validation and connection - setup for direct httpx clients. - """ + """Resolve and validate *host* at TCP-connect time; return dialable IP strings. Closes the + DNS-rebinding gap between pre-flight validation and connect for direct httpx clients. The + result is capped at ``_MAX_SSRF_CONNECT_IPS``, but EVERY answer is validated.""" hostname = _normalize_hostname(host) if not hostname: raise SSRFConnectionBlocked("Blocked request with empty hostname") - if hostname in _BLOCKED_HOSTNAMES: raise SSRFConnectionBlocked(f"Blocked request to internal hostname: {hostname}") - allow_private = _global_allow_private_urls() or _allows_private_ip_resolution(hostname, scheme) - try: - addr_info = socket.getaddrinfo(hostname, port, socket.AF_UNSPEC, socket.SOCK_STREAM) + addr_info = _getaddrinfo(hostname, port) except socket.gaierror as exc: raise SSRFConnectionBlocked(f"Blocked request - DNS resolution failed for: {hostname}") from exc - safe_ips: list[str] = [] - seen: set[str] = set() for raw, ip_str, ip in _iter_resolved_ips(addr_info): if ip is None: raise SSRFConnectionBlocked( f"Blocked request - unparseable IP address {raw!r} for hostname {hostname}" ) from ValueError(f"{ip_str!r} does not appear to be an IPv4 or IPv6 address") - reason = _resolved_ip_block_reason(ip, allow_private) if reason is not None: - raise SSRFConnectionBlocked( - f"Blocked request to {reason} during connect: {hostname} -> {ip_str}" - ) - - if ip_str not in seen and len(safe_ips) < _MAX_SSRF_CONNECT_IPS: + raise SSRFConnectionBlocked(f"Blocked request to {reason} during connect: {hostname} -> {ip_str}") + if ip_str not in safe_ips and len(safe_ips) < _MAX_SSRF_CONNECT_IPS: safe_ips.append(ip_str) - seen.add(ip_str) - if not safe_ips: raise SSRFConnectionBlocked(f"Blocked request - DNS returned no results for: {hostname}") - return safe_ips # capped at _MAX_SSRF_CONNECT_IPS, but EVERY answer above was validated + return safe_ips class _SSRFGuardedBackendBase: """httpcore backend that re-resolves + validates at connect time and dials a vetted IP. - - Host/SNI stay on the original hostname (the transport still sees ``host``); - Unix sockets are refused outright. Candidate IPs are tried in order and the - last connect error is re-raised so callers see the real network failure. - """ + Host/SNI stay on the original hostname; Unix sockets are refused outright. Candidate IPs are + tried in order and the last connect error is re-raised so callers see the real failure.""" def __init__(self, backend: Any, schemes_by_origin_var: Any): self._backend = backend self._schemes_by_origin_var = schemes_by_origin_var - def _connect_scheme(self, host: str, port: int) -> str: - return _safe_connect_scheme(host, port, self._schemes_by_origin_var.get({})) + def _connect_ips(self, host: str, port: int) -> list[str]: + scheme = self._schemes_by_origin_var.get({}).get((host, port)) or ("https" if port == 443 else "http") + return _resolved_http_connect_ips(host, port, scheme) + + @staticmethod + def _connect_errors() -> tuple: + import httpcore + return (httpcore.ConnectError, httpcore.ConnectTimeout) @staticmethod def _no_usable_ips(host: str, last_exc: Exception | None) -> Exception: @@ -479,33 +369,23 @@ class _SSRFGuardedBackendBase: return last_exc return SSRFConnectionBlocked(f"Blocked request - DNS returned no usable IPs for: {host}") + def connect_unix_socket(self, path: str, timeout: float | None = None, socket_options: Any = None) -> Any: + raise SSRFConnectionBlocked("Blocked Unix socket connection in SSRF-safe transport") + class _SSRFGuardedAsyncNetworkBackend(_SSRFGuardedBackendBase): def __init__(self, schemes_by_origin_var: Any): from httpcore._backends.auto import AutoBackend - super().__init__(AutoBackend(), schemes_by_origin_var) - async def connect_tcp( - self, - host: str, - port: int, - timeout: float | None = None, - local_address: str | None = None, - socket_options: Any = None, - ) -> Any: - import httpcore - - scheme = self._connect_scheme(host, port) - ips = await asyncio.to_thread(_resolved_http_connect_ips, host, port, scheme) - + async def connect_tcp(self, host: str, port: int, timeout: float | None = None, + local_address: str | None = None, socket_options: Any = None) -> Any: last_exc: Exception | None = None - for ip in ips: + for ip in await asyncio.to_thread(self._connect_ips, host, port): try: return await self._backend.connect_tcp( - ip, port, timeout=timeout, local_address=local_address, socket_options=socket_options, - ) - except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc: + ip, port, timeout=timeout, local_address=local_address, socket_options=socket_options) + except self._connect_errors() as exc: last_exc = exc raise self._no_usable_ips(host, last_exc) @@ -519,83 +399,62 @@ class _SSRFGuardedAsyncNetworkBackend(_SSRFGuardedBackendBase): class _SSRFGuardedNetworkBackend(_SSRFGuardedBackendBase): def __init__(self, schemes_by_origin_var: Any): from httpcore._backends.sync import SyncBackend - super().__init__(SyncBackend(), schemes_by_origin_var) - def connect_tcp( - self, - host: str, - port: int, - timeout: float | None = None, - local_address: str | None = None, - socket_options: Any = None, - ) -> Any: - import httpcore - - ips = _resolved_http_connect_ips(host, port, self._connect_scheme(host, port)) - + def connect_tcp(self, host: str, port: int, timeout: float | None = None, + local_address: str | None = None, socket_options: Any = None) -> Any: last_exc: Exception | None = None - for ip in ips: + for ip in self._connect_ips(host, port): try: return self._backend.connect_tcp( - ip, port, timeout=timeout, local_address=local_address, socket_options=socket_options, - ) - except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc: + ip, port, timeout=timeout, local_address=local_address, socket_options=socket_options) + except self._connect_errors() as exc: last_exc = exc raise self._no_usable_ips(host, last_exc) - def connect_unix_socket(self, path: str, timeout: float | None = None, socket_options: Any = None) -> Any: - raise SSRFConnectionBlocked("Blocked Unix socket connection in SSRF-safe transport") - def sleep(self, seconds: float) -> None: self._backend.sleep(seconds) def _origin_scheme_context(request: Any) -> dict[tuple[str, int], str]: - host = request.url.host - port = request.url.port - scheme = request.url.scheme - if not host or port is None or scheme not in _HTTP_SCHEMES: - return {} - return {(host, port): scheme} + host, port, scheme = request.url.host, request.url.port, request.url.scheme + return {(host, port): scheme} if host and port is not None and scheme in _HTTP_SCHEMES else {} + + +@contextmanager +def _origin_scope(schemes_by_origin_var: Any, request: Any): + """Expose the request's origin scheme to the connect-time backend for this request only.""" + token = schemes_by_origin_var.set(_origin_scheme_context(request)) + try: + yield + finally: + schemes_by_origin_var.reset(token) def _install_ssrf_guard_on_transport(transport: Any, schemes_by_origin_var: Any, *, is_async: bool = False) -> None: - """Swap the transport's pool network backend for the SSRF-guarded one (idempotent). - - Only the client's direct transport is guarded; proxy mounts are left alone so - final-target resolution is delegated to the (trusted) proxy egress. - """ + """Swap the transport's pool network backend for the SSRF-guarded one (idempotent). Only the + direct transport is guarded; proxy mounts delegate final-target resolution to the trusted proxy.""" state = getattr(transport, "__dict__", {}) if transport is not None else {} if transport is None or state.get("_hermes_ssrf_guarded", False): return - label = "async httpx transport" if is_async else "httpx transport" pool = state.get("_pool") if pool is None or not hasattr(pool, "_network_backend"): raise SSRFConnectionBlocked(f"Unsupported {label} cannot be made SSRF-safe") backend_cls = _SSRFGuardedAsyncNetworkBackend if is_async else _SSRFGuardedNetworkBackend pool._network_backend = backend_cls(schemes_by_origin_var) - method_name = "handle_async_request" if is_async else "handle_request" handle = getattr(transport, method_name, None) if handle is None: raise SSRFConnectionBlocked(f"Unsupported {label} cannot be made SSRF-safe") async def guarded_async(request: Any) -> Any: - token = schemes_by_origin_var.set(_origin_scheme_context(request)) - try: + with _origin_scope(schemes_by_origin_var, request): return await handle(request) - finally: - schemes_by_origin_var.reset(token) def guarded_sync(request: Any) -> Any: - token = schemes_by_origin_var.set(_origin_scheme_context(request)) - try: + with _origin_scope(schemes_by_origin_var, request): return handle(request) - finally: - schemes_by_origin_var.reset(token) - setattr(transport, method_name, guarded_async if is_async else guarded_sync) transport._hermes_ssrf_guarded = True @@ -603,24 +462,16 @@ def _install_ssrf_guard_on_transport(transport: Any, schemes_by_origin_var: Any, def _install_ssrf_guard_on_client(client: Any, *, is_async: bool = False) -> None: """Guard ``client._transport`` only; ``_mounts`` (env/explicit proxies) stay untouched.""" import contextvars - var_name = "hermes_ssrf_async_origin_schemes" if is_async else "hermes_ssrf_origin_schemes" _install_ssrf_guard_on_transport( - getattr(client, "__dict__", {}).get("_transport"), - contextvars.ContextVar(var_name), - is_async=is_async, - ) + getattr(client, "__dict__", {}).get("_transport"), contextvars.ContextVar(var_name), is_async=is_async) def create_ssrf_safe_async_client(**kwargs: Any) -> Any: - """Create an ``httpx.AsyncClient`` with connect-time SSRF validation. - - Direct HTTP(S) connections are resolved, validated, and dialed by IP at - TCP-connect time while the request hostname is preserved for Host, SNI, and - certificate verification. Proxied requests delegate resolution to the proxy. - """ + """``httpx.AsyncClient`` with connect-time SSRF validation: direct HTTP(S) connections are + resolved, validated, and dialed by IP while the hostname is preserved for Host, SNI, and + certificate verification. Proxied requests delegate resolution to the proxy.""" import httpx - client = httpx.AsyncClient(**kwargs) _install_ssrf_guard_on_client(client, is_async=True) return client @@ -629,30 +480,19 @@ def create_ssrf_safe_async_client(**kwargs: Any) -> Any: def create_ssrf_safe_client(**kwargs: Any) -> Any: """Create an ``httpx.Client`` with connect-time SSRF validation.""" import httpx - client = httpx.Client(**kwargs) _install_ssrf_guard_on_client(client) return client def redirect_target_from_response(response: Any) -> Optional[str]: - """Return the redirect target visible from inside an httpx response hook. - - ``response.next_request`` is frequently ``None`` inside hooks (populated later - by the redirect follower), which would make an SSRF redirect guard silently - never fire. Resolve from the ``Location`` header first (relative via - ``urljoin``), falling back to ``next_request``. - """ + """Redirect target visible from inside an httpx response hook. ``response.next_request`` is + frequently ``None`` inside hooks (populated later by the follower), which would make an SSRF + redirect guard silently never fire — so resolve from ``Location`` first, then ``next_request``.""" if not getattr(response, "is_redirect", False): return None - - headers = getattr(response, "headers", {}) or {} - location = headers.get("location") + location = (getattr(response, "headers", {}) or {}).get("location") if location: return urljoin(str(getattr(response, "url", "")), str(location)) - next_request = getattr(response, "next_request", None) - if next_request: - return str(next_request.url) - - return None + return str(next_request.url) if next_request else None diff --git a/tools/video_generation_tool.py b/tools/video_generation_tool.py index 287f21dfa3..1397c2ac57 100644 --- a/tools/video_generation_tool.py +++ b/tools/video_generation_tool.py @@ -3,9 +3,8 @@ (``agent/video_gen_provider.py`` ABC, ``agent/video_gen_registry.py``, ``plugins/video_gen//``). Ships **no in-tree provider**: enable a plugin and select it in ``hermes tools`` → Video -Generation. Covers text-, image- and reference-to-video; the tool layer only does lightweight -validation and each provider clamps/ignores unsupported params inside ``generate``, so the -surface stays stable as providers ship. Video edit/extend are deliberately not exposed here. +Generation. The tool layer only does lightweight validation; each provider clamps/ignores +unsupported params inside ``generate``. Video edit/extend are deliberately not exposed here. """ from __future__ import annotations @@ -19,8 +18,7 @@ from agent.video_gen_provider import ( COMMON_RESOLUTIONS, DEFAULT_ASPECT_RATIO, DEFAULT_RESOLUTION, - error_response, -) + error_response) from tools.registry import registry, tool_error logger = logging.getLogger(__name__) @@ -75,23 +73,15 @@ VIDEO_GENERATE_SCHEMA: Dict[str, Any] = { } -# --------------------------------------------------------------------------- -# Config readers (mirror image_generation_tool.py) -# --------------------------------------------------------------------------- def _read_video_gen_key(key: str) -> Optional[str]: """Return the stripped ``video_gen.`` string from config.yaml, or None.""" try: - from hermes_cli.config import load_config - - cfg = load_config() - section = cfg.get("video_gen") if isinstance(cfg, dict) else None - value = section.get(key) if isinstance(section, dict) else None + from hermes_cli.config import cfg_get, load_config + value = cfg_get(load_config(), "video_gen", key) except Exception as exc: logger.debug("Could not read video_gen config: %s", exc) return None - if isinstance(value, str) and value.strip(): - return value.strip() - return None + return value.strip() if isinstance(value, str) and value.strip() else None def _read_configured_video_provider() -> Optional[str]: @@ -106,32 +96,21 @@ def _discovered_registry(): """Import the provider registry after (idempotent) plugin discovery so user-installed plugins are visible.""" from agent import video_gen_registry from hermes_cli.plugins import _ensure_plugins_discovered - _ensure_plugins_discovered() return video_gen_registry, _ensure_plugins_discovered -# --------------------------------------------------------------------------- -# Availability check + provider resolution -# --------------------------------------------------------------------------- def check_video_generation_requirements() -> bool: """True when at least one registered provider reports available.""" try: registry_mod, _ = _discovered_registry() - for provider in registry_mod.list_providers(): - try: - if provider.is_available(): - return True - except Exception: - continue + return any(_provider_call(p, "is_available", False) for p in registry_mod.list_providers()) except Exception: - pass - return False + return False def _resolve_active_provider(): - """Active provider or None; forces a discovery refresh on a miss (long-lived sessions - that started before a plugin was installed).""" + """Active provider or None; a miss forces a discovery refresh (sessions older than the plugin install).""" try: registry_mod, ensure_discovered = _discovered_registry() provider = registry_mod.get_active_provider() @@ -147,32 +126,23 @@ def _resolve_active_provider(): def _missing_provider_error(configured: Optional[str]) -> str: if configured: return json.dumps(error_response( - error=( - f"video_gen.provider='{configured}' is set but no plugin " - f"registered that name. Run `hermes plugins list` to see " - f"installed video gen backends, or `hermes tools` → Video " - f"Generation to pick one." - ), - error_type="provider_not_registered", - provider=configured, - )) + error=(f"video_gen.provider='{configured}' is set but no plugin registered that name. " + f"Run `hermes plugins list` to see installed video gen backends, or " + f"`hermes tools` → Video Generation to pick one."), + error_type="provider_not_registered", provider=configured)) return json.dumps(error_response( - error=( - "No video generation backend is configured. Run `hermes tools` → " - "Video Generation to enable one (xAI, FAL, or Google Veo)." - ), - error_type="no_provider_configured", - )) + error=("No video generation backend is configured. Run `hermes tools` → " + "Video Generation to enable one (xAI, FAL, or Google Veo)."), + error_type="no_provider_configured")) + + +_BOOL_WORDS = {"true": True, "1": True, "yes": True, "on": True, + "false": False, "0": False, "no": False, "off": False} -# --------------------------------------------------------------------------- -# Handler -# --------------------------------------------------------------------------- def _coerce_int(value: Any) -> Optional[int]: - if value is None or value == "": - return None try: - return int(value) + return None if value is None or value == "" else int(value) except (TypeError, ValueError): return None @@ -180,21 +150,15 @@ def _coerce_int(value: Any) -> Optional[int]: def _coerce_bool(value: Any) -> Optional[bool]: if isinstance(value, bool): return value - if isinstance(value, str): - return {"true": True, "1": True, "yes": True, "on": True, - "false": False, "0": False, "no": False, "off": False}.get(value.strip().lower()) - return None + return _BOOL_WORDS.get(value.strip().lower()) if isinstance(value, str) else None def _normalize_reference_images(value: Any) -> Optional[List[str]]: - if value is None: - return None if isinstance(value, str): value = [value] if not isinstance(value, (list, tuple)): return None - out = [item.strip() for item in value if isinstance(item, str) and item.strip()] - return out or None + return [item.strip() for item in value if isinstance(item, str) and item.strip()] or None def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str: @@ -205,29 +169,28 @@ def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str: # Confinement chokepoint (mirrors image_generate): non-local backends hand providers data: URLs. from tools.image_generation_tool import _confine_source_images - image_url, reference_image_urls, confine_error = _confine_source_images( image_url, reference_image_urls, task_id) if confine_error is not None: return confine_error - duration = _coerce_int(args.get("duration")) - aspect_ratio = (args.get("aspect_ratio") or DEFAULT_ASPECT_RATIO).strip() or DEFAULT_ASPECT_RATIO - resolution = (args.get("resolution") or DEFAULT_RESOLUTION).strip() or DEFAULT_RESOLUTION - negative_prompt = (args.get("negative_prompt") or "").strip() or None - audio = _coerce_bool(args.get("audio")) - seed = _coerce_int(args.get("seed")) - upscale = _coerce_bool(args.get("upscale")) + # Coerced BEFORE validation (ordering parity: a bad value raises before a missing prompt). + optional = { + "duration": _coerce_int(args.get("duration")), + "aspect_ratio": (args.get("aspect_ratio") or DEFAULT_ASPECT_RATIO).strip() or DEFAULT_ASPECT_RATIO, + "resolution": (args.get("resolution") or DEFAULT_RESOLUTION).strip() or DEFAULT_RESOLUTION, + "negative_prompt": (args.get("negative_prompt") or "").strip() or None, + "audio": _coerce_bool(args.get("audio")), + "seed": _coerce_int(args.get("seed")), + "upscale": _coerce_bool(args.get("upscale"))} model_override = (args.get("model") or "").strip() or None - # Soft validation — providers do their own; a backend may accept image-only, our surface never does. + # Soft validation — providers do their own; our surface never accepts image-only. if not prompt: return tool_error("prompt is required for video generation") if "operation" in args or "video_url" in args: return tool_error( "video_generate only supports text-to-video, image-to-video, and " - "reference-to-video; use a provider-specific tool for video edit/extend" - ) - + "reference-to-video; use a provider-specific tool for video edit/extend") configured = _read_configured_video_provider() provider = _resolve_active_provider() if provider is None: @@ -235,20 +198,9 @@ def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str: # Explicit arg wins, then config, then provider default. model = model_override or _read_configured_video_model() or provider.default_model() - kwargs: Dict[str, Any] = { - "model": model, - "_model_override_explicit": bool(model_override), - "image_url": image_url, - "reference_image_urls": reference_image_urls, - "duration": duration, - "aspect_ratio": aspect_ratio, - "resolution": resolution, - "negative_prompt": negative_prompt, - "audio": audio, - "seed": seed, - "upscale": upscale, - } + "model": model, "_model_override_explicit": bool(model_override), + "image_url": image_url, "reference_image_urls": reference_image_urls, **optional} # Drop None entries so providers see clean defaults. kwargs = {k: v for k, v in kwargs.items() if v is not None} pname = getattr(provider, "name", "?") @@ -256,42 +208,28 @@ def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str: def _err(error: str, error_type: str) -> str: return json.dumps(error_response( error=error, error_type=error_type, - provider=getattr(provider, "name", ""), model=model or "", prompt=prompt, - )) - + provider=getattr(provider, "name", ""), model=model or "", prompt=prompt)) try: result = provider.generate(prompt=prompt, **kwargs) except TypeError as exc: # An un-widened provider signature is a plugin bug, not a caller error. - logger.warning( - "video_gen provider '%s' rejected kwargs (signature too narrow): %s", - pname, exc, - ) + logger.warning("video_gen provider '%s' rejected kwargs (signature too narrow): %s", pname, exc) return _err( - f"Provider '{pname}' signature is " - f"out of date with the video_generate schema. Report this " - f"to the plugin author.", - "provider_contract", - ) + f"Provider '{pname}' signature is out of date with the video_generate schema. " + f"Report this to the plugin author.", + "provider_contract") except Exception as exc: logger.warning("video_gen provider '%s' raised: %s", pname, exc) return _err(f"Provider '{pname}' error: {exc}", "provider_exception") - if not isinstance(result, dict): return _err("Provider returned a non-dict result", "provider_contract") - return json.dumps(result) -# --------------------------------------------------------------------------- -# Dynamic schema — reflect the active backend's actual capabilities -# --------------------------------------------------------------------------- -# Surfacing the per-model surface (modalities, enums, durations, audio/negative-prompt) -# means the model usually gets the call right first try. model_tools.get_tool_definitions() -# keys its cache on config.yaml mtime, so the schema rebuilds on provider/model change. - -# Optional params advertised only when the provider's capabilities() sets the flag -# (order = schema property order). +# Dynamic schema — reflects the active backend's actual capabilities so the model usually gets +# the call right first try. model_tools.get_tool_definitions() keys its cache on config.yaml +# mtime, so the schema rebuilds on provider/model change. Optional params below are advertised +# only when the provider's capabilities() sets the flag (order = schema property order). _CAPABILITY_PARAMS = ( ("supports_negative_prompt", "negative_prompt", { "type": "string", @@ -334,34 +272,31 @@ _GENERIC_DESCRIPTION = ( def _schema(description: str, properties: Dict[str, Any]) -> Dict[str, Any]: return { "description": description, - "parameters": {"type": "object", "properties": properties, "required": ["prompt"]}, - } + "parameters": {"type": "object", "properties": properties, "required": ["prompt"]}} + + +def _provider_call(provider: Any, method: str, default: Any) -> Any: + """``provider.()`` or ``default`` when it raises or returns a falsy value.""" + try: + return getattr(provider, method)() or default + except Exception: + return default def _build_dynamic_video_schema() -> Dict[str, Any]: - """Render description AND params from capabilities() + the model's catalog entry; enums and - duration bounds tighten to the active model. Unadvertised args are still accepted (replay compat).""" + """Description AND params from capabilities() + the model's catalog entry; enums and duration + bounds tighten to the active model. Unadvertised args are still accepted (replay compat).""" static_props = VIDEO_GENERATE_SCHEMA["parameters"]["properties"] parts: List[str] = [_GENERIC_DESCRIPTION] - configured_model = _read_configured_video_model() provider = _resolve_active_provider() - if provider is None: parts.append( "\nNo video backend is available. Calls will return an error " - "until the user picks one via `hermes tools` → Video Generation." - ) + "until the user picks one via `hermes tools` → Video Generation.") return _schema("\n".join(parts), {"prompt": static_props["prompt"]}) - - try: - caps = provider.capabilities() or {} - except Exception: - caps = {} - try: - models = provider.list_models() or [] - except Exception: - models = [] + caps = _provider_call(provider, "capabilities", {}) + models = _provider_call(provider, "list_models", []) active_model = configured_model or provider.default_model() model_meta = next((m for m in models if isinstance(m, dict) and m.get("id") == active_model), {}) @@ -374,11 +309,9 @@ def _build_dynamic_video_schema() -> Dict[str, Any]: if "image" in model_modalities and "text" not in model_modalities: parts.append( "- this model is image-to-video only — image_url is REQUIRED; " - "text-only calls will be rejected" - ) + "text-only calls will be rejected") elif "text" in model_modalities and "image" not in model_modalities: parts.append("- this model is text-to-video only — image_url is not supported") - effective_modalities = model_modalities or set(caps.get("modalities") or []) can_i2v = "image" in effective_modalities t2v = "text" in effective_modalities @@ -386,33 +319,26 @@ def _build_dynamic_video_schema() -> Dict[str, Any]: parts.append("- image-to-video only: image_url is REQUIRED") elif not can_i2v: parts.append("- text-to-video only (no image input)") - if provider.name == "xai": parts.append( "- chaining: for edit/extend pass the public HTTPS MP4 in `video` " "or `public_url` from the prior Imagine result (files-cdn). For " "image-to-video / reference-to-video pass public image URLs the " - "same way" - ) + "same way") try: from tools.xai_http import xai_storage_notice_text - notice = xai_storage_notice_text("video_gen") except Exception: notice = "" if notice: parts.append(f"- storage: {notice}") - properties: Dict[str, Any] = {"prompt": static_props["prompt"]} - if can_i2v: properties["image_url"] = { "type": "string", "description": ( "Public HTTPS URL of a still image to animate " - "(image-to-video). Omit for text-to-video." - ), - } + "(image-to-video). Omit for text-to-video.")} max_refs = int(caps.get("max_reference_images") or 0) if max_refs > 0: properties["reference_image_urls"] = { @@ -421,10 +347,7 @@ def _build_dynamic_video_schema() -> Dict[str, Any]: "maxItems": max_refs, "description": ( f"Up to {max_refs} public HTTPS reference image URLs " - "(style or character refs)." - ), - } - + "(style or character refs).")} min_duration = model_meta.get("min_duration", caps.get("min_duration")) max_duration = model_meta.get("max_duration", caps.get("max_duration")) duration_param = dict(static_props["duration"]) @@ -433,8 +356,7 @@ def _build_dynamic_video_schema() -> Dict[str, Any]: duration_param["maximum"] = int(max_duration) duration_param["description"] = ( f"Video duration in seconds ({min_duration}-{max_duration}). " - "Omit for the provider default." - ) + "Omit for the provider default.") properties["duration"] = duration_param # Tighten enums to the active backend's actual sets when declared. @@ -443,7 +365,6 @@ def _build_dynamic_video_schema() -> Dict[str, Any]: if caps.get(caps_key): param["enum"] = list(caps[caps_key]) properties[key] = param - for flag, key, param in _CAPABILITY_PARAMS: if caps.get(flag): properties[key] = param @@ -451,8 +372,7 @@ def _build_dynamic_video_schema() -> Dict[str, Any]: parts.append( "- audio: native stereo audio is generated with every video " "(always on; no toggle) — describe the desired sound in the " - "prompt" - ) + "prompt") properties["model"] = static_props["model"] return _schema("\n".join(parts), properties) @@ -466,5 +386,4 @@ registry.register( requires_env=[], is_async=False, emoji="🎬", - dynamic_schema_overrides=_build_dynamic_video_schema, -) + dynamic_schema_overrides=_build_dynamic_video_schema) diff --git a/tools/vision_tools.py b/tools/vision_tools.py index 50dea3a1fc..b8a45eddff 100644 --- a/tools/vision_tools.py +++ b/tools/vision_tools.py @@ -1,10 +1,9 @@ #!/usr/bin/env python3 """Vision tools: ``vision_analyze`` (image) and ``video_analyze``. -Images resolve through :mod:`tools.image_source`, are normalized to a -provider-supported format (:mod:`tools.vision_tools_image_prep`), then either -attach natively to a vision-capable main model (multimodal tool-result -envelope) or are described by the auxiliary vision LLM router. +Images resolve through :mod:`tools.image_source`, are normalized to a provider-supported format +(:mod:`tools.vision_tools_image_prep`), then either attach natively to a vision-capable main +model (multimodal tool-result envelope) or are described by the auxiliary vision LLM router. """ import base64 @@ -16,13 +15,12 @@ import logging import os import uuid from pathlib import Path -from typing import Any, Awaitable, Dict, Optional +from typing import Any, Awaitable, Callable, Dict, NamedTuple, Optional from urllib.parse import urlparse import httpx -# ``agent.auxiliary_client`` costs ~50 ms cold (credential_pool → auth → rich); -# only the handlers need it. Both names stay module attributes so tests can -# patch ``tools.vision_tools.async_call_llm``; truthy-skip means injected mocks win. +# ``agent.auxiliary_client`` costs ~50 ms cold; only the handlers need it. Both names stay +# module attributes so tests can patch ``tools.vision_tools.async_call_llm`` (truthy-skip: mocks win). async_call_llm: Any = None extract_content_or_reasoning: Any = None @@ -30,14 +28,9 @@ extract_content_or_reasoning: Any = None def _load_auxiliary_client() -> None: global async_call_llm, extract_content_or_reasoning if async_call_llm is None or extract_content_or_reasoning is None: - from agent.auxiliary_client import ( - async_call_llm as _acl, - extract_content_or_reasoning as _ecr, - ) - if async_call_llm is None: - async_call_llm = _acl - if extract_content_or_reasoning is None: - extract_content_or_reasoning = _ecr + from agent import auxiliary_client as _aux + async_call_llm = async_call_llm or _aux.async_call_llm + extract_content_or_reasoning = extract_content_or_reasoning or _aux.extract_content_or_reasoning from hermes_constants import get_hermes_dir @@ -54,70 +47,50 @@ from tools.vision_tools_image_prep import ( # noqa: F401 — re-exported for te _normalize_to_supported_image, _rasterize_svg_to_png, _supported_media_types, - _validate_raster_image_decodable, -) + _validate_raster_image_decodable) logger = logging.getLogger(__name__) _debug = DebugSession("vision_tools", env_var="VISION_TOOLS_DEBUG") -def _read_vision_setting(env_var: str, key: str, cast, minimum=None): - """Env var → config.yaml ``auxiliary.vision.`` → None. +def _cfg_auxiliary(*keys: str, default=None): + """``auxiliary.`` from config.yaml; ``default`` when config is unavailable.""" + try: + from hermes_cli.config import cfg_get, load_config + return cfg_get(load_config(), "auxiliary", *keys, default=default) + except Exception: + return default - Values that fail ``cast`` or fall below ``minimum`` are skipped in favor of - the next source (a cap can never be disabled by a bad value). - """ + +def _read_vision_setting(env_var: str, key: str, cast, minimum=None): + """Env var → ``auxiliary.vision.`` → None. Values that fail ``cast`` or fall below + ``minimum`` are skipped in favor of the next source (a cap can never be disabled by a bad value).""" def _accept(raw): try: val = cast(raw) except (TypeError, ValueError): return None return val if minimum is None or val >= minimum else None - - env_val = os.getenv(env_var, "").strip() - if env_val: - val = _accept(env_val) - if val is not None: - return val - try: - from hermes_cli.config import cfg_get, load_config - raw = cfg_get(load_config(), "auxiliary", "vision", key) - if raw is not None: - return _accept(raw) - except Exception: - pass - return None + val = _accept(os.getenv(env_var, "").strip() or None) + return val if val is not None else _accept(_cfg_auxiliary("vision", key)) -def _resolve_download_timeout() -> float: - """HTTP download timeout (separate from ``auxiliary.vision.timeout``, which governs the LLM call).""" - val = _read_vision_setting("HERMES_VISION_DOWNLOAD_TIMEOUT", "download_timeout", float) - return 30.0 if val is None else val +# HTTP download timeout (separate from ``auxiliary.vision.timeout``, which governs the LLM call). +_VISION_DOWNLOAD_TIMEOUT = _read_vision_setting("HERMES_VISION_DOWNLOAD_TIMEOUT", "download_timeout", float) +if _VISION_DOWNLOAD_TIMEOUT is None: + _VISION_DOWNLOAD_TIMEOUT = 30.0 - -_VISION_DOWNLOAD_TIMEOUT = _resolve_download_timeout() - -# Hard cap on downloaded media (50 MB): bounds memory/disk against -# attacker-hosted multi-gigabyte files. +# Hard cap on downloaded media (50 MB): bounds memory/disk against attacker-hosted files. _VISION_MAX_DOWNLOAD_BYTES = 50 * 1024 * 1024 -# --------------------------------------------------------------------------- -# CPU-burst concurrency cap (vision encode/resize) -# --------------------------------------------------------------------------- -# A turn can fan out dozens of vision_analyze calls ("analyze every frame"); -# each does a CPU-heavy base64 encode + Pillow resize. Sessions share one -# process, so unbounded encodes saturate every core and starve the shared event -# loop (the dashboard liveness probe flapped UNHEALTHY in prod). We cap ONLY the -# CPU burst — the LLM calls stay fully concurrent — on a dedicated executor sized -# to the usable core count (the resource actually exhausted; no fixed ceiling). -# It must be a threading primitive: each call runs via model_tools._run_async on -# a PER-THREAD event loop, so an asyncio semaphore cannot coordinate across them. -# The default executor is NOT used: it is shared with the gateway/web server. -import threading # noqa: F401 (kept for downstream importers / patch targets) - - +# CPU-burst cap: a turn can fan out dozens of vision_analyze calls (base64 encode + Pillow +# resize each) that would saturate every core and starve the shared event loop. Only the CPU +# burst is capped (LLM calls stay concurrent), on a dedicated executor sized to usable cores. +# Must be a threading primitive — each call runs via model_tools._run_async on a PER-THREAD +# loop, so an asyncio semaphore cannot coordinate; the default executor is shared with the +# gateway/web server and is deliberately NOT used. def _detect_host_cpus() -> int: """Usable CPU count (``sched_getaffinity`` honors cpuset pinning), at least 1.""" try: @@ -127,16 +100,15 @@ def _detect_host_cpus() -> int: def _resolve_vision_cpu_workers() -> int: - """HERMES_VISION_MAX_CONCURRENCY → ``auxiliary.vision.max_concurrency`` → host cores (values < 1 ignored).""" + """HERMES_VISION_MAX_CONCURRENCY → ``auxiliary.vision.max_concurrency`` → host cores (< 1 ignored).""" val = _read_vision_setting("HERMES_VISION_MAX_CONCURRENCY", "max_concurrency", int, minimum=1) - return _detect_host_cpus() if val is None else val + return val or _detect_host_cpus() _VISION_CPU_WORKERS = _resolve_vision_cpu_workers() _vision_cpu_executor = ThreadPoolExecutor( - max_workers=_VISION_CPU_WORKERS, - thread_name_prefix="vision-encode", + max_workers=_VISION_CPU_WORKERS, thread_name_prefix="vision-encode", ) @@ -144,38 +116,25 @@ async def _run_encode_on_cpu_executor(fn, *args, **kwargs): """Run a sync encode/resize callable on the bounded vision CPU executor (never the LLM call).""" import functools loop = asyncio.get_running_loop() - return await loop.run_in_executor( - _vision_cpu_executor, functools.partial(fn, *args, **kwargs) - ) - - -def _image_url_shape_ok(url: str) -> bool: - """HTTP(S) shape check only (scheme + netloc; no DNS). Extension-less CDN URLs pass.""" - if not url or not isinstance(url, str) or not url.startswith(("http://", "https://")): - return False - return bool(urlparse(url).netloc) + return await loop.run_in_executor(_vision_cpu_executor, functools.partial(fn, *args, **kwargs)) async def _validate_image_url_async(url: str) -> bool: - """Validate remote image URL (SSRF guard) without blocking the event loop on DNS.""" - if not _image_url_shape_ok(url): + """HTTP(S) shape check (extension-less CDN URLs pass) + SSRF guard, DNS off the event loop.""" + if not (isinstance(url, str) and url.startswith(("http://", "https://")) and urlparse(url).netloc): return False from tools.url_safety import async_is_safe_url return await async_is_safe_url(url) def _is_retryable_download_error(error: Exception) -> bool: - """True only for transient download failures worth retrying. - - Fail-fast: 4xx other than 429 (missing/forbidden), PermissionError (policy - or SSRF block), ValueError (too large / blocked redirect — deterministic). - Retryable: 429, 5xx, transport errors and anything unclassified. - """ + """Transient failures only: 429, 5xx, transport errors and anything unclassified. Fail-fast on + other 4xx, PermissionError (policy/SSRF block) and ValueError (too large / blocked redirect).""" if isinstance(error, (PermissionError, ValueError)): return False if isinstance(error, httpx.HTTPStatusError): status = error.response.status_code - return not (400 <= status < 500 and status != 429) + return status == 429 or not 400 <= status < 500 return True @@ -185,102 +144,23 @@ _DOWNLOAD_USER_AGENT = ( ) -async def _stream_download_to_file( - client, - url: str, - destination: Path, - max_bytes: int, - *, - headers: dict, - media_label: str = "Image", -) -> Path: - """Stream a GET to *destination* via a temp file with a running size cap. - - The body is never fully buffered: chunks go to a temp file and the running - count is checked after each one (Content-Length gives an early reject but - servers can omit or lie, so the streaming cap is authoritative). The temp - file is atomically moved onto *destination* on success, deleted on failure. - """ - from utils import atomic_replace - - async with client.stream("GET", url, headers=headers) as response: - response.raise_for_status() - - cl = response.headers.get("content-length") - if cl: - try: - declared_size = int(cl) - except ValueError: - declared_size = None - if declared_size is not None and declared_size > max_bytes: - raise ValueError( - f"{media_label} too large ({declared_size} bytes, max {max_bytes})" - ) - - blocked = check_website_access(str(response.url)) - if blocked: - raise PermissionError(blocked["message"]) - - tmp_destination = destination.with_name( - f".{destination.name}.{uuid.uuid4().hex}.tmp" - ) - bytes_written = 0 - try: - with tmp_destination.open("wb") as f: - async for chunk in response.aiter_bytes(): - if not chunk: - continue - bytes_written += len(chunk) - if bytes_written > max_bytes: - raise ValueError( - f"{media_label} too large ({bytes_written} bytes, max {max_bytes})" - ) - f.write(chunk) - atomic_replace(tmp_destination, destination) - except Exception: - try: - tmp_destination.unlink(missing_ok=True) - except OSError: - logger.debug( - "Could not delete partial download: %s", tmp_destination, exc_info=True - ) - raise - - return destination - - async def _ssrf_redirect_guard(response): - """Re-validate each redirect target: a public URL that 302s to - http://169.254.169.254/ would otherwise bypass the pre-flight is_safe_url check. - Async because httpx.AsyncClient awaits event hooks.""" + """Re-validate each redirect target (a public URL 302ing to 169.254.169.254 would otherwise + bypass the pre-flight check). Async because httpx.AsyncClient awaits hooks.""" from tools.url_safety import async_is_safe_url, redirect_target_from_response redirect_url = redirect_target_from_response(response) if redirect_url and not await async_is_safe_url(redirect_url): - raise ValueError( - f"Blocked redirect to private/internal address: {redirect_url}" - ) + raise ValueError(f"Blocked redirect to private/internal address: {redirect_url}") async def _download_media( - url: str, - destination: Path, - max_retries: int, - *, - media_label: str, - accept: str, - max_bytes: int, - timeout: float, - retry_all: bool, -) -> Path: - """Shared SSRF-safe streaming download with exponential backoff (2s/4s/8s). - - ``retry_all=False`` (images) only retries transient errors per - :func:`_is_retryable_download_error` — a 404/403 never succeeds on retry, so - burning three backoff rounds just inflates latency. ``retry_all=True`` - (video) keeps the legacy retry-everything behavior. - """ + url: str, destination: Path, max_retries: int, *, + media_label: str, accept: str, max_bytes: int, timeout: float, retry_all: bool) -> Path: + """SSRF-safe streaming download with exponential backoff (2s/4s/8s). ``retry_all=False`` (images) + retries only :func:`_is_retryable_download_error` errors — a 404/403 never succeeds on retry.""" + from utils import atomic_replace + from tools.url_safety import create_ssrf_safe_async_client destination.parent.mkdir(parents=True, exist_ok=True) - last_error = None for attempt in range(max_retries): try: @@ -288,29 +168,55 @@ async def _download_media( if blocked: raise PermissionError(blocked["message"]) - from tools.url_safety import create_ssrf_safe_async_client - # follow_redirects for CDNs; the client validates DNS at connect # time and the hook re-validates each redirect target. async with create_ssrf_safe_async_client( - timeout=timeout, - follow_redirects=True, - event_hooks={"response": [_ssrf_redirect_guard]}, - ) as client: - await _stream_download_to_file( - client, url, destination, max_bytes, - headers={"User-Agent": _DOWNLOAD_USER_AGENT, "Accept": accept}, - media_label=media_label, - ) + timeout=timeout, follow_redirects=True, event_hooks={"response": [_ssrf_redirect_guard]}, + ) as client, client.stream( + "GET", url, headers={"User-Agent": _DOWNLOAD_USER_AGENT, "Accept": accept}, + ) as response: + response.raise_for_status() + + # Content-Length gives an early reject but servers can omit or lie, + # so the streaming cap below is authoritative. + cl = response.headers.get("content-length") + try: + declared_size = int(cl) if cl else None + except ValueError: + declared_size = None + if declared_size is not None and declared_size > max_bytes: + raise ValueError(f"{media_label} too large ({declared_size} bytes, max {max_bytes})") + blocked = check_website_access(str(response.url)) + if blocked: + raise PermissionError(blocked["message"]) + + # Stream to a temp file, atomically moved onto destination on success. + tmp_destination = destination.with_name(f".{destination.name}.{uuid.uuid4().hex}.tmp") + bytes_written = 0 + try: + with tmp_destination.open("wb") as f: + async for chunk in response.aiter_bytes(): + if not chunk: + continue + bytes_written += len(chunk) + if bytes_written > max_bytes: + raise ValueError( + f"{media_label} too large ({bytes_written} bytes, max {max_bytes})" + ) + f.write(chunk) + atomic_replace(tmp_destination, destination) + except Exception: + try: + tmp_destination.unlink(missing_ok=True) + except OSError: + logger.debug("Could not delete partial download: %s", tmp_destination, exc_info=True) + raise return destination except Exception as e: last_error = e - final = attempt >= max_retries - 1 - if final or (not retry_all and not _is_retryable_download_error(e)): - logger.error( - "%s download failed after %s attempt(s): %s", - media_label, attempt + 1, str(e)[:100], exc_info=True, - ) + if attempt >= max_retries - 1 or (not retry_all and not _is_retryable_download_error(e)): + logger.error("%s download failed after %s attempt(s): %s", + media_label, attempt + 1, str(e)[:100], exc_info=True) if not retry_all: raise break @@ -320,13 +226,11 @@ async def _download_media( if not retry_all: logger.warning("Retrying in %ss...", wait_time) await asyncio.sleep(wait_time) - # Reaching here means max_retries was non-positive (or video exhausted retries). if last_error is not None: raise last_error raise RuntimeError( - f"_download_{media_label.lower()} exited retry loop without attempting (max_retries={max_retries})" - ) + f"_download_{media_label.lower()} exited retry loop without attempting (max_retries={max_retries})") async def _download_image(image_url: str, destination: Path, max_retries: int = 3) -> Path: @@ -334,26 +238,21 @@ async def _download_image(image_url: str, destination: Path, max_retries: int = return await _download_media( image_url, destination, max_retries, media_label="Image", accept="image/*,*/*;q=0.8", - max_bytes=_VISION_MAX_DOWNLOAD_BYTES, timeout=_VISION_DOWNLOAD_TIMEOUT, - retry_all=False, - ) + max_bytes=_VISION_MAX_DOWNLOAD_BYTES, timeout=_VISION_DOWNLOAD_TIMEOUT, retry_all=False) def _image_to_base64_data_url(image_path: Path, mime_type: Optional[str] = None) -> str: """``data:;base64,...`` for a file (MIME from extension when not given).""" - encoded = base64.b64encode(image_path.read_bytes()).decode("ascii") - return f"data:{mime_type or _determine_mime_type(image_path)};base64,{encoded}" + mime = mime_type or _determine_mime_type(image_path) + return f"data:{mime};base64,{base64.b64encode(image_path.read_bytes()).decode('ascii')}" # Absolute hard ceiling for vision payloads (20 MB): no major provider accepts more. _MAX_BASE64_BYTES = 20 * 1024 * 1024 -# Proactive embed caps for conversation-history reuse. Native vision_analyze -# bakes the data URL into the tool result, re-sent every later turn; a 4 MB / -# 7900px embed cost ~100-260K billed tokens per image. 256 KB keeps a 1568px -# screenshot cheap enough to ride the session; Anthropic's tokenizer downsamples -# to a 1568px long edge anyway, so pixels past that cost wire bytes for no -# fidelity. The 20 MB / provider 5 MB caps remain as one-shot safety nets. +# Proactive embed caps for history reuse: the native path bakes the data URL into the tool +# result, re-sent every later turn (a 4 MB embed cost ~100-260K billed tokens). Anthropic +# downsamples to a 1568px long edge anyway, so pixels past that cost wire bytes for no fidelity. _EMBED_TARGET_BYTES = 256 * 1024 _EMBED_MAX_DIMENSION = 1568 @@ -372,62 +271,40 @@ def _is_image_size_error(error: Exception) -> bool: return any(hint in err_str for hint in _SIZE_ERROR_HINTS + ("image_url", "invalid_request")) -def _build_scale_note( - scale_info: Optional[dict], - crop_offset: Optional[dict], -) -> Optional[str]: - """Coordinate-mapping disclosure for downscale (``scale_info``) and/or region - crop (``crop_offset``); ``None`` when neither applied — no note, no noise.""" +def _build_scale_note(scale_info: Optional[dict], crop_offset: Optional[dict]) -> Optional[str]: + """Coordinate-mapping disclosure for downscale and/or region crop; ``None`` when neither applied.""" parts = [] if scale_info: ow, oh = scale_info["orig_width"], scale_info["orig_height"] nw, nh = scale_info["new_width"], scale_info["new_height"] - fx = ow / nw if nw else 1.0 - fy = oh / nh if nh else 1.0 - if f"{fx:.2f}" == f"{fy:.2f}": - factor_clause = ( - f"multiply any coordinates you report by {fx:.2f} " - f"to map back to the original image." - ) - else: - factor_clause = ( - f"multiply any x coordinates you report by {fx:.2f} and " - f"any y coordinates by {fy:.2f} to map back to the " - f"original image." - ) - parts.append( - f"Image downscaled from {ow}x{oh} to {nw}x{nh} for vision; " - f"{factor_clause}" - ) + fx, fy = (ow / nw if nw else 1.0), (oh / nh if nh else 1.0) + axes = (f"any coordinates you report by {fx:.2f}" if f"{fx:.2f}" == f"{fy:.2f}" + else f"any x coordinates you report by {fx:.2f} and any y coordinates by {fy:.2f}") + parts.append(f"Image downscaled from {ow}x{oh} to {nw}x{nh} for vision; " + f"multiply {axes} to map back to the original image.") if crop_offset: parts.append( f"Analysis was performed on a cropped region of the original " f"image starting at offset ({crop_offset['x']}, " f"{crop_offset['y']}); coordinates are relative to that crop " - f"origin — add the offset to map back to the full image." - ) + f"origin — add the offset to map back to the full image.") return " ".join(parts) if parts else None def _import_pillow_for_resize(): - """Pillow is a lazy-installable soft dependency; return ``PIL.Image`` or None. - - ``prompt=False``: a blocking input() deadlocks the interactive CLI where - prompt_toolkit owns stdin. The install is gated by - security.allow_lazy_installs, so reaching it is already opt-in. - """ + """Return ``PIL.Image`` or None (Pillow is a lazy-installable soft dependency). ``prompt=False``: + a blocking input() deadlocks the CLI where prompt_toolkit owns stdin; the install is already + gated by security.allow_lazy_installs.""" try: from PIL import Image - return Image except ImportError: - pass - try: - from tools.lazy_deps import ensure as _ensure_dep - _ensure_dep("tool.vision", prompt=False) - from PIL import Image - return Image - except Exception: - return None + try: + from tools.lazy_deps import ensure as _ensure_dep + _ensure_dep("tool.vision", prompt=False) + from PIL import Image + except Exception: + return None + return Image def _resize_image_for_vision(image_path: Path, mime_type: Optional[str] = None, @@ -437,71 +314,55 @@ def _resize_image_for_vision(image_path: Path, mime_type: Optional[str] = None, force_jpeg: bool = False) -> str: """Base64 data URL, progressively downscaled with Pillow while over budget. - Halves dimensions (aspect-preserving, 64px floor) up to 4 times; JPEG also - walks a quality ladder (85/70/50) at each step. Without Pillow, or if it - still doesn't fit, returns the best attempt (or raw bytes) and lets the - caller apply the size check. - - ``max_dimension``: force a downscale above this long edge even when bytes - fit (Anthropic's 8000px cap is independent of bytes). ``force_jpeg``: - re-encode PNG input as JPEG when a resize is needed — PNG's only shrink - lever is halving dimensions, which destroys text legibility on dense - screenshots; history-reuse embeds opt in. Images under both caps return unchanged. + Halves dimensions (aspect-preserving, 64px floor) up to 4 times; JPEG also walks a + quality ladder (85/70/50) per step. Without Pillow, or if it still doesn't fit, returns + the best attempt (or raw bytes) and lets the caller apply the size check. + ``max_dimension``: force a downscale above this long edge even when bytes fit + (Anthropic's 8000px cap is independent of bytes). ``force_jpeg``: re-encode PNG as JPEG + when resizing — halving PNG dimensions destroys text legibility on dense screenshots. """ file_size = image_path.stat().st_size estimated_b64 = (file_size * 4) // 3 + 100 # base64 ~4/3 + data URL header - needs_resize_for_bytes = estimated_b64 > max_base64_bytes - needs_resize_for_dims = ( - max_dimension is not None and _image_exceeds_dimension(image_path, max_dimension) - ) - data_url = None - if not needs_resize_for_bytes and not needs_resize_for_dims: + if estimated_b64 <= max_base64_bytes and not ( + max_dimension is not None and _image_exceeds_dimension(image_path, max_dimension) + ): data_url = _image_to_base64_data_url(image_path, mime_type=mime_type) if len(data_url) <= max_base64_bytes: return data_url def _raw() -> str: return data_url or _image_to_base64_data_url(image_path, mime_type=mime_type) - Image = _import_pillow_for_resize() if Image is None: logger.info("Pillow not installed — cannot auto-resize oversized image") return _raw() # caller will raise the size error - logger.info("Image file is %.1f MB (estimated base64 %.1f MB, limit %.1f MB, max_dimension=%s), auto-resizing...", file_size / (1024 * 1024), estimated_b64 / (1024 * 1024), max_base64_bytes / (1024 * 1024), max_dimension) - - mime = mime_type or _determine_mime_type(image_path) # JPEG for photos (smaller), PNG for transparency — unless force_jpeg. - pil_format = "PNG" if (mime == "image/png" and not force_jpeg) else "JPEG" - out_mime = "image/png" if pil_format == "PNG" else "image/jpeg" - + is_png = (mime_type or _determine_mime_type(image_path)) == "image/png" and not force_jpeg + pil_format, out_mime = ("PNG", "image/png") if is_png else ("JPEG", "image/jpeg") try: img = Image.open(image_path) except Exception as exc: logger.info("Pillow cannot open image for resizing: %s", exc) return _raw() - # JPEG cannot encode alpha/palette modes (force_jpeg routes PNGs here). - if pil_format == "JPEG" and img.mode not in {"RGB", "L"}: - img = img.convert("RGB") - quality_steps = (85, 70, 50) if pil_format == "JPEG" else (None,) + # JPEG cannot encode alpha/palette modes (force_jpeg routes PNGs here). + if not is_png and img.mode not in {"RGB", "L"}: + img = img.convert("RGB") + quality_steps = (None,) if is_png else (85, 70, 50) orig_dims = prev_dims = (img.width, img.height) candidate = None def _record_scale(w: int, h: int) -> None: if scale_out is not None and (w, h) != orig_dims: - scale_out.update(orig_width=orig_dims[0], orig_height=orig_dims[1], - new_width=w, new_height=h) - + scale_out.update(orig_width=orig_dims[0], orig_height=orig_dims[1], new_width=w, new_height=h) for attempt in range(5): if attempt > 0: - # Halve, then re-derive the scale from whichever axis hit the 64px - # floor so both axes shrink by the same factor. - new_w = max(int(img.width * 0.5), 64) - new_h = max(int(img.height * 0.5), 64) + # Halve, then re-derive from whichever axis hit the 64px floor so both shrink equally. + new_w, new_h = max(int(img.width * 0.5), 64), max(int(img.height * 0.5), 64) if new_w == 64 and img.width > 0: new_h = max(int(img.height * (64 / img.width)), 64) elif new_h == 64 and img.height > 0: @@ -511,40 +372,29 @@ def _resize_image_for_vision(image_path: Path, mime_type: Optional[str] = None, img = img.resize((new_w, new_h), Image.LANCZOS) prev_dims = (new_w, new_h) logger.info("Resized to %dx%d (attempt %d)", new_w, new_h, attempt) - + dims_ok = max_dimension is None or max(img.width, img.height) <= max_dimension for q in quality_steps: buf = BytesIO() img.save(buf, format=pil_format, **({} if q is None else {"quality": q})) candidate = f"data:{out_mime};base64,{base64.b64encode(buf.getvalue()).decode('ascii')}" - dims_ok = max_dimension is None or max(img.width, img.height) <= max_dimension if len(candidate) <= max_base64_bytes and dims_ok: logger.info("Auto-resized image fits: %.1f MB (quality=%s, %dx%d)", - len(candidate) / (1024 * 1024), q, - img.width, img.height) + len(candidate) / (1024 * 1024), q, img.width, img.height) _record_scale(img.width, img.height) return candidate - - if candidate is not None: - logger.warning("Auto-resize could not fit image under %.1f MB (best: %.1f MB)", - max_base64_bytes / (1024 * 1024), len(candidate) / (1024 * 1024)) - _record_scale(img.width, img.height) - return candidate - return _raw() + if candidate is None: + return _raw() + logger.warning("Auto-resize could not fit image under %.1f MB (best: %.1f MB)", + max_base64_bytes / (1024 * 1024), len(candidate) / (1024 * 1024)) + _record_scale(img.width, img.height) + return candidate -# --------------------------------------------------------------------------- -# Native fast path: when the active main model supports vision, skip the aux -# LLM and return the image bytes as a multimodal tool-result envelope. The -# agent loop unwraps it into an OpenAI-style content list on the `tool` role; -# provider adapters translate that per backend, so the main model "sees" the -# pixels directly on its next turn. -# --------------------------------------------------------------------------- - -# Providers whose tool results accept image content (spec docs verified -# Apr-2026): Anthropic Messages (+ aggregators proxying Claude — assume support, -# falling back to text would regress their frontier models), OpenAI Chat -# Completions / Responses. Gemini is gated on model: only 3.x supports -# multimodal functionResponse. +# Native fast path: a vision-capable main model gets the image bytes as a multimodal +# tool-result envelope (the agent loop unwraps it into an OpenAI-style content list on the +# `tool` role); no aux LLM, no information loss. Providers whose tool results accept image +# content: Anthropic Messages (and aggregators proxying Claude — assume support), OpenAI +# Chat/Responses. Gemini is gated on model: only 3.x supports multimodal functionResponse. _TOOL_RESULT_MEDIA_PROVIDERS = frozenset({ "openrouter", "nous", "vertex", "bedrock", "anthropic-vertex", "google-vertex", "anthropic", "claude", "anthropic-direct", @@ -554,42 +404,33 @@ _GEMINI_PROVIDERS = frozenset({"google", "gemini", "google-gemini", "google-vert def _supports_media_in_tool_results(provider: str, model: str) -> bool: - """Whether provider+model accepts image content inside a tool-result message. - - Unknown providers are conservatively False (caller falls back to the aux-LLM - text path) unless their ``ProviderProfile`` declares ``supports_vision``. - """ - if not isinstance(provider, str): - return False - p = provider.strip().lower() + """Whether provider+model accepts image content inside a tool-result message. Unknown + providers are False (caller falls back to aux-LLM text) unless their ``ProviderProfile`` + declares ``supports_vision``.""" + p = provider.strip().lower() if isinstance(provider, str) else "" if not p: return False if p in _TOOL_RESULT_MEDIA_PROVIDERS: return True if p in _GEMINI_PROVIDERS: - if not isinstance(model, str): - return False - m = model.strip().lower() + m = model.strip().lower() if isinstance(model, str) else "" return any(tag in m for tag in ("gemini-3", "gemini-pro-3", "gemini-flash-3")) try: from providers import get_provider_profile profile = get_provider_profile(p) - if profile is not None and profile.supports_vision: - return True + return profile is not None and bool(profile.supports_vision) except Exception: - pass - return False + return False def _should_use_native_vision_fast_path() -> bool: - """True when image routing resolves to ``native`` AND the provider accepts - images in tool results, or the user set the ``model.supports_vision`` - override (escape hatch for custom/local providers). Any failure → False.""" + """True when image routing resolves to ``native`` AND the provider accepts images in tool + results, or the user set the ``model.supports_vision`` override (escape hatch for + custom/local providers). Any failure → False.""" try: from agent.auxiliary_client import _read_main_provider, _read_main_model from agent.image_routing import decide_image_input_mode, _lookup_supports_vision from hermes_cli.config import load_config - provider = _read_main_provider() model = _read_main_model() cfg = load_config() @@ -597,61 +438,40 @@ def _should_use_native_vision_fast_path() -> bool: return False return ( _supports_media_in_tool_results(provider, model) - or _lookup_supports_vision(provider, model, cfg) is True - ) + or _lookup_supports_vision(provider, model, cfg) is True) except Exception as exc: logger.debug("Native vision fast-path check failed: %s", exc) return False def _build_native_vision_tool_result( - image_url: str, - question: str, - image_data_url: str, - image_size_bytes: int, + image_url: str, question: str, image_data_url: str, image_size_bytes: int, scale_note: Optional[str] = None, ) -> Dict[str, Any]: - """Multimodal tool-result envelope (``_multimodal`` + ``content`` list). - - The text part is intentionally minimal — the model already has the question - in context; it acknowledges the image is visible. ``text_summary`` is the - fallback for providers without multimodal tool results. - """ + """Multimodal tool-result envelope. The text part is intentionally minimal (the model already + has the question); ``text_summary`` is the fallback for providers without multimodal tool results.""" text_part = ( "Image loaded into your context — you can see it natively now. " - "Use your built-in vision to answer the user." - ) + "Use your built-in vision to answer the user.") if isinstance(question, str) and question.strip(): text_part += f"\n\nQuestion: {question.strip()}" if scale_note: text_part += f"\n\nNote: {scale_note}" - - summary = ( - f"Image attached natively for the main model " - f"({image_size_bytes / 1024:.1f} KB). " - "Answer using built-in vision." - ) - return { "_multimodal": True, "content": [ {"type": "text", "text": text_part}, - {"type": "image_url", "image_url": {"url": image_data_url}}, - ], - "text_summary": summary, - "meta": { - "image_url": image_url[:200], - "size_bytes": image_size_bytes, - "native_vision": True, - }, - } + {"type": "image_url", "image_url": {"url": image_data_url}}], + "text_summary": ( + f"Image attached natively for the main model ({image_size_bytes / 1024:.1f} KB). " + "Answer using built-in vision."), + "meta": {"image_url": image_url[:200], "size_bytes": image_size_bytes, "native_vision": True}} def _unlink_quietly(path: Optional[Path]) -> None: if path is not None: try: - if path.exists(): - path.unlink() + path.unlink(missing_ok=True) except Exception: pass @@ -660,71 +480,52 @@ class _ImagePrepError(ValueError): """Raised by :func:`_prepare_image`; the message is user-facing.""" -class _PreparedImage: +class _PreparedImage(NamedTuple): """Temp image ready to encode; ``path`` is owned by the caller (delete it).""" - __slots__ = ("path", "mime", "size_bytes", "crop_offset") - - def __init__(self, path: Path, mime: Optional[str], size_bytes: int, crop_offset: dict): - self.path, self.mime, self.size_bytes, self.crop_offset = path, mime, size_bytes, crop_offset + path: Path + mime: Optional[str] + size_bytes: int + crop_offset: dict async def _prepare_image( image_url: str, task_id: Optional[str], region: Optional[list], *, validate_decode: bool, ) -> _PreparedImage: """Resolve → materialize → normalize → (validate) → (crop). Raises ``_ImagePrepError``. - - The single resolver unifies data:/http/file/local/container sources and - enforces terminal-backend confinement; bytes land in a temp file so the - path-based encode/resize pipeline is reused. Unsupported formats (SVG, BMP) - are converted to PNG BEFORE encoding — an unsupported media_type baked into - immutable history would 400 on every resume. The crop runs BEFORE any - downscale so the region keeps the full resolution budget. Blocking - rasterizer/Pillow work is offloaded. Intermediate temp files are deleted; - on error nothing is left behind. - """ + Unsupported formats (SVG, BMP) become PNG BEFORE encoding — an unsupported media_type baked + into immutable history would 400 on every resume. The crop runs BEFORE any downscale so the + region keeps the full resolution budget. On error no temp file is left.""" from tools.image_source import ImageResolutionError, ResolveContext, resolve_image_source - try: resolved = await resolve_image_source(image_url, ResolveContext(task_id=task_id)) except ImageResolutionError as exc: raise _ImagePrepError(str(exc)) from exc - temp_dir = get_hermes_dir("cache/vision", "temp_vision_images") temp_dir.mkdir(parents=True, exist_ok=True) path = temp_dir / f"temp_image_{uuid.uuid4()}.img" await asyncio.to_thread(path.write_bytes, resolved.data) - mime = resolved.mime - size_bytes = len(resolved.data) - crop_offset: dict = {} + mime, size_bytes, crop_offset = resolved.mime, len(resolved.data), {} try: - normalized_path, mime, norm_err = await asyncio.to_thread( - _normalize_to_supported_image, path, mime, - ) + normalized_path, mime, norm_err = await asyncio.to_thread(_normalize_to_supported_image, path, mime) if norm_err or normalized_path is None: raise _ImagePrepError(norm_err or "Image normalization failed.") if normalized_path != path: _unlink_quietly(path) path = normalized_path size_bytes = path.stat().st_size - if validate_decode: decode_error = await _run_encode_on_cpu_executor( _validate_raster_image_decodable, path, - _VISION_MAX_VALIDATED_FRAME_COUNT, _VISION_MAX_VALIDATED_AGGREGATE_PIXELS, - ) + _VISION_MAX_VALIDATED_FRAME_COUNT, _VISION_MAX_VALIDATED_AGGREGATE_PIXELS) if decode_error: raise _ImagePrepError(decode_error) - if region is not None: cropped_path, cropped_mime, crop_err = await asyncio.to_thread( - _crop_image_region, path, region, offset_out=crop_offset, - ) + _crop_image_region, path, region, offset_out=crop_offset) if crop_err or cropped_path is None: raise _ImagePrepError(crop_err or "Region crop failed.") _unlink_quietly(path) - path = cropped_path - mime = cropped_mime - size_bytes = path.stat().st_size + path, mime, size_bytes = cropped_path, cropped_mime, cropped_path.stat().st_size except BaseException: _unlink_quietly(path) raise @@ -733,74 +534,52 @@ async def _prepare_image( def _too_large_message(image_data_url: str) -> str: return ( - f"Image too large for vision API: base64 payload is " - f"{len(image_data_url) / (1024 * 1024):.1f} MB " - f"(limit {_MAX_BASE64_BYTES / (1024 * 1024):.0f} MB) " - f"even after resizing. Install Pillow " - f"(`pip install Pillow`) for better auto-resize, " - f"or compress the image manually." + f"Image too large for vision API: base64 payload is {len(image_data_url) / (1024 * 1024):.1f} MB " + f"(limit {_MAX_BASE64_BYTES / (1024 * 1024):.0f} MB) even after resizing. Install Pillow " + f"(`pip install Pillow`) for better auto-resize, or compress the image manually.") + + +async def _resize_prepared(prepared: _PreparedImage, scale_info: dict, **kwargs) -> str: + """Run :func:`_resize_image_for_vision` on the CPU executor for a prepared image.""" + return await _run_encode_on_cpu_executor( + _resize_image_for_vision, prepared.path, mime_type=prepared.mime, scale_out=scale_info, **kwargs, ) async def _vision_analyze_native( - image_url: str, - question: str, - task_id: Optional[str] = None, - region: Optional[list] = None, + image_url: str, question: str, task_id: Optional[str] = None, region: Optional[list] = None, ) -> Any: - """Fast path for vision-capable main models. - - Returns a ``_multimodal`` envelope dict on success, or a JSON error string - (the normal tool-result contract) on failure. - """ + """Fast path for vision-capable main models: a ``_multimodal`` envelope dict on success, + or a JSON error string (the normal tool-result contract) on failure.""" if not isinstance(image_url, str) or not image_url.strip(): return tool_error("image_url is required", success=False) - prepared: Optional[_PreparedImage] = None try: from tools.interrupt import is_interrupted if is_interrupted(): return tool_error("Interrupted", success=False) - try: prepared = await _prepare_image(image_url, task_id, region, validate_decode=True) except _ImagePrepError as exc: return tool_error(str(exc), success=False) - image_data_url = await _run_encode_on_cpu_executor( - _image_to_base64_data_url, prepared.path, mime_type=prepared.mime, - ) - - # Proactive embed cap: this image is re-sent on every later turn, so - # resize DOWN to the history-reuse target whenever the byte or long-edge - # cap is exceeded, not just at the 20 MB hard ceiling. + _image_to_base64_data_url, prepared.path, mime_type=prepared.mime) + # Proactive embed cap: this image is re-sent on every later turn, so resize DOWN to the + # history-reuse target whenever the byte or long-edge cap is exceeded, not just at 20 MB. _scale_info: dict = {} _over_dims = await _run_encode_on_cpu_executor( - _image_exceeds_dimension, prepared.path, _EMBED_MAX_DIMENSION, - ) + _image_exceeds_dimension, prepared.path, _EMBED_MAX_DIMENSION) if len(image_data_url) > _EMBED_TARGET_BYTES or _over_dims: - image_data_url = await _run_encode_on_cpu_executor( - _resize_image_for_vision, - prepared.path, mime_type=prepared.mime, - max_base64_bytes=_EMBED_TARGET_BYTES, - max_dimension=_EMBED_MAX_DIMENSION, - scale_out=_scale_info, - force_jpeg=True, - ) + image_data_url = await _resize_prepared( + prepared, _scale_info, + max_base64_bytes=_EMBED_TARGET_BYTES, max_dimension=_EMBED_MAX_DIMENSION, force_jpeg=True) # Reject rather than embed a session-wedging payload. if len(image_data_url) > _MAX_BASE64_BYTES: return tool_error(_too_large_message(image_data_url), success=False) - return _build_native_vision_tool_result( - image_url=image_url, - question=question, - image_data_url=image_data_url, + image_url=image_url, question=question, image_data_url=image_data_url, image_size_bytes=prepared.size_bytes, - scale_note=_build_scale_note( - _scale_info or None, prepared.crop_offset or None, - ), - ) - + scale_note=_build_scale_note(_scale_info or None, prepared.crop_offset or None)) except Exception as exc: logger.warning("Native vision fast path failed: %s", exc) return tool_error(f"Native vision failed: {exc}", success=False) @@ -810,29 +589,31 @@ async def _vision_analyze_native( _unlink_quietly(prepared.path) -def _read_vision_call_settings(default_timeout: float, *, min_timeout: Optional[float] = None): - """``auxiliary.vision.timeout`` / ``.temperature`` from config.yaml (defaults 120s-ish / 0.1). - - Local vision models (llama.cpp, ollama) can take well over 30s, hence the - generous defaults; ``min_timeout`` lets video enforce a floor. - """ +def _aux_call_kwargs(messages: list, model: Optional[str], default_timeout: float, *, + min_timeout: Optional[float] = None) -> dict: + """``async_call_llm`` kwargs with ``auxiliary.vision.timeout`` / ``.temperature`` from config. + Local vision models (llama.cpp, ollama) can take well over 30s, hence generous defaults + (temperature 0.1); ``min_timeout`` lets video enforce a floor.""" timeout, temperature = default_timeout, 0.1 try: - from hermes_cli.config import cfg_get, load_config - _vision_cfg = cfg_get(load_config(), "auxiliary", "vision", default={}) - _vt = _vision_cfg.get("timeout") - if _vt is not None: - timeout = float(_vt) if min_timeout is None else max(float(_vt), min_timeout) - _vtemp = _vision_cfg.get("temperature") - if _vtemp is not None: - temperature = float(_vtemp) + _vision_cfg = _cfg_auxiliary("vision", default={}) or {} + if _vision_cfg.get("timeout") is not None: + timeout = max(float(_vision_cfg["timeout"]), float("-inf") if min_timeout is None else min_timeout) + if _vision_cfg.get("temperature") is not None: + temperature = float(_vision_cfg["temperature"]) except Exception: pass - return timeout, temperature + return {"task": "vision", "messages": messages, "temperature": temperature, "timeout": timeout, + **({"model": model} if model else {})} -# Error-message classification for the aux-LLM paths: first matching hint set -# wins (order matters — billing before capability before size/format). +def _media_messages(user_prompt: str, part_type: str, data_url: str) -> list: + """Single user message: text + one ``image_url``/``video_url`` data-URL part.""" + return [{"role": "user", "content": [ + {"type": "text", "text": user_prompt}, {"type": part_type, part_type: {"url": data_url}}]}] + + +# Aux-LLM error classification: first matching hint set wins (billing → capability → size/format). _BILLING_HINTS = ("402", "insufficient", "payment required", "credits", "billing") _IMAGE_ERROR_RULES = ( (_BILLING_HINTS, @@ -859,215 +640,145 @@ _VIDEO_ERROR_RULES = ( "The video is too large for the API. Try compressing or trimming " "the video (max ~50 MB). Error: {e}"), ) - - -def _classify_analysis_error(e: Exception, rules, fallback: str, **fmt) -> str: - err_str = str(e).lower() - for hints, template in rules: - if any(hint in err_str for hint in hints): - return template.format(e=e, **fmt) - return fallback.format(e=e) +# kind -> (debug tool name, error rules) +_ANALYSIS_KINDS = { + "image": ("vision_analyze_tool", _IMAGE_ERROR_RULES), + "video": ("video_analyze_tool", _VIDEO_ERROR_RULES)} def _debug_call_data(kind: str, source: str, user_prompt: str, model) -> dict: + prompt = user_prompt[:200] + "..." if len(user_prompt) > 200 else user_prompt return { - "parameters": { - f"{kind}_url": source, - "user_prompt": user_prompt[:200] + "..." if len(user_prompt) > 200 else user_prompt, - "model": model, - }, - "error": None, - "success": False, - "analysis_length": 0, - "model_used": model, - f"{kind}_size_bytes": 0, - } + "parameters": {f"{kind}_url": source, "user_prompt": prompt, "model": model}, + "error": None, "success": False, "analysis_length": 0, "model_used": model, f"{kind}_size_bytes": 0} -def _finish_analysis(tool_name: str, debug_call_data: dict, result: dict) -> str: - _debug.log_call(tool_name, debug_call_data) - _debug.save() - return json.dumps(result, indent=2, ensure_ascii=False) - - -def _cleanup_temp_media(path: Optional[Path], label: str) -> None: - if path and path.exists(): - try: - path.unlink() - logger.debug("Cleaned up temporary %s file", label) - except Exception as cleanup_error: - logger.warning( - "Could not delete temporary file: %s", cleanup_error, exc_info=True - ) - - -async def _call_vision_llm(call_kwargs: dict, empty_log: str): - """Call the aux vision LLM, retrying once on empty content (reasoning-only response).""" +async def _call_vision_llm(call_kwargs: dict, empty_log: str, response=None): + """Aux vision LLM analysis text, retrying once on empty content (reasoning-only response). + ``response``: an already-made first call (the image path handles size-error retries itself).""" _load_auxiliary_client() - response = await async_call_llm(**call_kwargs) + if response is None: + response = await async_call_llm(**call_kwargs) analysis = extract_content_or_reasoning(response) if not analysis: logger.warning(empty_log) - response = await async_call_llm(**call_kwargs) - analysis = extract_content_or_reasoning(response) + analysis = extract_content_or_reasoning(await async_call_llm(**call_kwargs)) return analysis -async def vision_analyze_tool( - image_url: str, - user_prompt: str, - model: str = None, - task_id: Optional[str] = None, - region: Optional[list] = None, -) -> str: - """Describe an image (URL, local path, data: URL) with the auxiliary vision LLM. +async def _run_analysis( + kind: str, source: str, user_prompt: str, model: Optional[str], + stage: Callable[[str, dict, list], Awaitable[tuple]]) -> str: + """Aux-LLM analysis skeleton for image/video: interrupt check → ``stage`` → JSON result. - ``user_prompt`` is pre-formatted by the caller. Returns JSON - ``{"success": bool, "analysis": str}`` (``analysis`` carries the error - explanation on failure). Temp images live under $HERMES_HOME/cache/vision/. + ``stage(user_prompt, debug_call_data, temp_paths)`` returns ``(analysis, scale_note)``; temp + files it appends to ``temp_paths`` are deleted in ``finally``. Returns JSON + ``{"success": bool, "analysis": str}`` (``analysis`` carries the error explanation on failure). """ + tool_name, rules = _ANALYSIS_KINDS[kind] if not isinstance(user_prompt, str): user_prompt = str(user_prompt) if user_prompt is not None else "" - debug_call_data = _debug_call_data("image", image_url, user_prompt, model) + debug_call_data = _debug_call_data(kind, source, user_prompt, model) + temp_paths: list = [] + + def finish(result: dict) -> str: + _debug.log_call(tool_name, debug_call_data) + _debug.save() + return json.dumps(result, indent=2, ensure_ascii=False) - prepared: Optional[_PreparedImage] = None try: from tools.interrupt import is_interrupted if is_interrupted(): return tool_error("Interrupted", success=False) - - logger.info("Analyzing image: %s", image_url[:60]) + logger.info("Analyzing %s: %s", kind, source[:60]) logger.info("User prompt: %s", user_prompt[:100]) + analysis, scale_note = await stage(user_prompt, debug_call_data, temp_paths) + analysis_length = len(analysis) if analysis else 0 + logger.info("%s analysis completed (%s characters)", kind.capitalize(), analysis_length) + analysis = analysis or f"There was a problem with the request and the {kind} could not be analyzed." + result = {"success": True, "analysis": f"[{scale_note}] {analysis}" if scale_note else analysis} + if scale_note: + result["scale_note"] = scale_note + debug_call_data.update(success=True, analysis_length=analysis_length) + return finish(result) + except Exception as e: + error_msg = f"Error analyzing {kind}: {str(e)}" + logger.error("%s", error_msg, exc_info=True) + err_str = str(e).lower() + template = next( + (tpl for hints, tpl in rules if any(hint in err_str for hint in hints)), + f"There was a problem with the request and the {kind} could not be analyzed. Error: {{e}}") + analysis = template.format(e=e, model=model) + debug_call_data["error"] = error_msg + return finish({"success": False, "error": error_msg, "analysis": analysis}) + finally: + for path in temp_paths: + if path and path.exists(): + try: + path.unlink() + logger.debug("Cleaned up temporary %s file", kind) + except Exception as cleanup_error: + logger.warning("Could not delete temporary file: %s", cleanup_error, exc_info=True) + +async def vision_analyze_tool( + image_url: str, user_prompt: str, model: str = None, + task_id: Optional[str] = None, region: Optional[list] = None) -> str: + """Describe an image (URL, local path, data: URL) with the auxiliary vision LLM. ``user_prompt`` + is pre-formatted by the caller. Temp images live under $HERMES_HOME/cache/vision/.""" + async def stage(prompt: str, debug_call_data: dict, temp_paths: list) -> tuple: prepared = await _prepare_image(image_url, task_id, region, validate_decode=False) + temp_paths.append(prepared.path) logger.info("Image ready (%.1f KB)", prepared.size_bytes / 1024) - # Send at full resolution first; on a size rejection, downscale and retry. logger.info("Converting image to base64...") image_data_url = await _run_encode_on_cpu_executor( _image_to_base64_data_url, prepared.path, mime_type=prepared.mime) logger.info("Image converted to base64 (%.1f KB)", len(image_data_url) / 1024) - _scale_info: dict = {} if len(image_data_url) > _MAX_BASE64_BYTES: - image_data_url = await _run_encode_on_cpu_executor( - _resize_image_for_vision, - prepared.path, mime_type=prepared.mime, - scale_out=_scale_info) + image_data_url = await _resize_prepared(prepared, _scale_info) if len(image_data_url) > _MAX_BASE64_BYTES: raise ValueError(_too_large_message(image_data_url)) - debug_call_data["image_size_bytes"] = prepared.size_bytes - - messages = [{ - "role": "user", - "content": [ - {"type": "text", "text": user_prompt}, - {"type": "image_url", "image_url": {"url": image_data_url}}, - ], - }] - + messages = _media_messages(prompt, "image_url", image_data_url) logger.info("Processing image with vision model...") - vision_timeout, vision_temperature = _read_vision_call_settings(120.0) - call_kwargs = { - "task": "vision", - "messages": messages, - "temperature": vision_temperature, - "timeout": vision_timeout, - } - if model: - call_kwargs["model"] = model + call_kwargs = _aux_call_kwargs(messages, model, 120.0) _load_auxiliary_client() try: response = await async_call_llm(**call_kwargs) except Exception as _api_err: - if (_is_image_size_error(_api_err) - and len(image_data_url) > _RESIZE_TARGET_BYTES): - logger.info( - "API rejected image (%.1f MB, likely too large); " - "auto-resizing to ~%.0f MB and retrying...", - len(image_data_url) / (1024 * 1024), - _RESIZE_TARGET_BYTES / (1024 * 1024), - ) - image_data_url = await _run_encode_on_cpu_executor( - _resize_image_for_vision, - prepared.path, mime_type=prepared.mime, - scale_out=_scale_info) - messages[0]["content"][1]["image_url"]["url"] = image_data_url - response = await async_call_llm(**call_kwargs) - else: + if not (_is_image_size_error(_api_err) and len(image_data_url) > _RESIZE_TARGET_BYTES): raise - - analysis = extract_content_or_reasoning(response) - if not analysis: - logger.warning("Vision LLM returned empty content, retrying once") + logger.info( + "API rejected image (%.1f MB, likely too large); auto-resizing to ~%.0f MB and retrying...", + len(image_data_url) / (1024 * 1024), _RESIZE_TARGET_BYTES / (1024 * 1024)) + image_data_url = await _resize_prepared(prepared, _scale_info) + messages[0]["content"][1]["image_url"]["url"] = image_data_url response = await async_call_llm(**call_kwargs) - analysis = extract_content_or_reasoning(response) - - analysis_length = len(analysis) - logger.info("Image analysis completed (%s characters)", analysis_length) - - analysis = analysis or "There was a problem with the request and the image could not be analyzed." - scale_note = _build_scale_note(_scale_info or None, prepared.crop_offset or None) - result = { - "success": True, - "analysis": f"[{scale_note}] {analysis}" if scale_note else analysis, - } - if scale_note: - result["scale_note"] = scale_note - - debug_call_data["success"] = True - debug_call_data["analysis_length"] = analysis_length - return _finish_analysis("vision_analyze_tool", debug_call_data, result) - - except Exception as e: - error_msg = f"Error analyzing image: {str(e)}" - logger.error("%s", error_msg, exc_info=True) - analysis = _classify_analysis_error( - e, _IMAGE_ERROR_RULES, - "There was a problem with the request and the image could not " - "be analyzed. Error: {e}", - model=model, - ) - debug_call_data["error"] = error_msg - return _finish_analysis("vision_analyze_tool", debug_call_data, { - "success": False, - "error": error_msg, - "analysis": analysis, - }) - - finally: - if prepared is not None: - _cleanup_temp_media(prepared.path, "image") + analysis = await _call_vision_llm( + call_kwargs, "Vision LLM returned empty content, retrying once", response) + return analysis, _build_scale_note(_scale_info or None, prepared.crop_offset or None) + return await _run_analysis("image", image_url, user_prompt, model, stage) def check_vision_requirements() -> bool: """True when ``call_llm(task="vision")`` could resolve a client. - Mirrors its runtime fallback chain: explicit ``auxiliary.vision.provider``, - then the auto chain (main provider → openrouter → nous) — without the auto - step the tool would vanish whenever the explicit name was unresolvable. - Probe mode skips real SDK client construction (openai import + SSL setup) - on the tool-gating path; resolution policy is identical. + Mirrors its fallback chain: explicit ``auxiliary.vision.provider``, then auto (main + provider → openrouter → nous) — without the auto step the tool would vanish whenever + the explicit name was unresolvable. Probe mode skips real SDK client construction. """ try: from agent.auxiliary_client import aux_probe_mode, resolve_vision_provider_client - except ImportError: - return False - try: with aux_probe_mode(): - _provider, client, _model = resolve_vision_provider_client() - if client is not None: - return True - _provider, client, _model = resolve_vision_provider_client(provider="auto") - return client is not None + return any( + resolve_vision_provider_client(**kw)[1] is not None for kw in ({}, {"provider": "auto"}) + ) except Exception: return False -# --------------------------------------------------------------------------- -# Registry -# --------------------------------------------------------------------------- from tools.registry import registry, tool_error VISION_ANALYZE_SCHEMA = { @@ -1111,37 +822,22 @@ VISION_ANALYZE_SCHEMA = { def _configured_aux_model(sections: tuple, env_vars: tuple) -> Optional[str]: - """First non-empty ``auxiliary.
.model`` from config.yaml, else the - first non-empty env var (legacy override), else None.""" - try: - from hermes_cli.config import cfg_get, load_config - _cfg = load_config() - for section in sections: - _vmodel = cfg_get(_cfg, "auxiliary", section, "model") - if _vmodel: - model = str(_vmodel).strip() or None - if model: - return model - break - except Exception: - pass - for env_var in env_vars: - val = os.getenv(env_var, "").strip() - if val: - return val - return None + """First non-empty ``auxiliary.
.model`` from config.yaml, else the first non-empty + env var (legacy override), else None.""" + for section in sections: + _vmodel = _cfg_auxiliary(section, "model") + if _vmodel: + if str(_vmodel).strip(): + return str(_vmodel).strip() + break + return next((v for v in (os.getenv(e, "").strip() for e in env_vars) if v), None) async def _handle_vision_analyze(args: Dict[str, Any], **kw: Any) -> str: - image_url = args.get("image_url", "") - question = args.get("question", "") - region = args.get("region") + image_url, question, region = args.get("image_url", ""), args.get("question", ""), args.get("region") task_id = kw.get("task_id") - - # No concurrency gate around the whole analysis — the CPU burst is bounded - # inside the encode/resize step, so multi-image fan-out keeps full request - # concurrency. Native fast path: main model sees the pixels directly, no - # aux call, no information loss. + # No concurrency gate around the whole analysis — the CPU burst is bounded inside the + # encode/resize step, so multi-image fan-out keeps full request concurrency. if _should_use_native_vision_fast_path(): logger.info("vision_analyze: native fast path") return await _vision_analyze_native(image_url, question, task_id=task_id, region=region) @@ -1149,8 +845,7 @@ async def _handle_vision_analyze(args: Dict[str, Any], **kw: Any) -> str: # Legacy path: aux LLM describes the image and we return its text. full_prompt = ( "Fully describe and explain everything about this image, then answer the " - f"following question:\n\n{question}" - ) + f"following question:\n\n{question}") model = _configured_aux_model(("vision",), ("AUXILIARY_VISION_MODEL",)) return await vision_analyze_tool(image_url, full_prompt, model, task_id=task_id, region=region) @@ -1162,14 +857,10 @@ registry.register( handler=_handle_vision_analyze, check_fn=check_vision_requirements, is_async=True, - emoji="👁️", -) + emoji="👁️") -# --------------------------------------------------------------------------- -# Video Analysis Tool -# --------------------------------------------------------------------------- - +# --- video_analyze -------------------------------------------------------- # Extension → MIME. avi/mkv fall back to mp4. _VIDEO_MIME_TYPES = { ".mp4": "video/mp4", @@ -1191,182 +882,86 @@ def _detect_video_mime_type(video_path: Path) -> Optional[str]: def _unsupported_video_format(suffix: str) -> str: - return ( - f"Unsupported video format: '{suffix}'. " - f"Supported: {', '.join(sorted(_VIDEO_MIME_TYPES.keys()))}" - ) + return f"Unsupported video format: '{suffix}'. Supported: {', '.join(sorted(_VIDEO_MIME_TYPES.keys()))}" def _video_to_base64_data_url(video_path: Path, mime_type: Optional[str] = None) -> str: - encoded = base64.b64encode(video_path.read_bytes()).decode("ascii") - mime = mime_type or _VIDEO_MIME_TYPES.get(video_path.suffix.lower(), "video/mp4") - return f"data:{mime};base64,{encoded}" + mime = mime_type or _detect_video_mime_type(video_path) or "video/mp4" + return f"data:{mime};base64,{base64.b64encode(video_path.read_bytes()).decode('ascii')}" -def _is_path_like_video_source(value: str) -> bool: - lowered = (value or "").strip().lower() - return bool(lowered) and not lowered.startswith(("http://", "https://", "data:")) - - -async def _materialize_video_from_terminal_backend(video_source: str, task_id: Optional[str]) -> Path: - """Read a path via the shared media resolver into a local temp video file. - - ``permitted=("video",)`` gives terminal-backend video reads the exact - pipeline vision_analyze uses: media-cache host reads (gateway downloads live - on the host, not in the sandbox), bounded in-sandbox exec-read, lazy env - bring-up, the credential-read guard, and the 50MB ingest cap. - """ - from tools.image_source import ImageResolutionError, ResolveContext, resolve_image_source - - source = video_source - if source.startswith("file://"): - source = source[len("file://"):] - suffix = Path(source).suffix.lower() - if suffix not in _VIDEO_MIME_TYPES: - raise ValueError(_unsupported_video_format(suffix)) - - try: - resolved = await resolve_image_source( - video_source, ResolveContext(task_id=task_id), permitted=("video",) - ) - except ImageResolutionError as exc: - raise ValueError(f"Could not read video from terminal backend: {exc}") from exc - - temp_dir = get_hermes_dir("cache/video", "temp_video_files") - temp_dir.mkdir(parents=True, exist_ok=True) - temp_path = temp_dir / f"terminal_video_{uuid.uuid4()}{suffix}" - temp_path.write_bytes(resolved.data) - return temp_path - - -async def _download_video(video_url: str, destination: Path, max_retries: int = 3) -> Path: - """Download video with SSRF protection; every failure class is retried.""" - return await _download_media( - video_url, destination, max_retries, - media_label="Video", accept="video/*,*/*;q=0.8", - max_bytes=_MAX_VIDEO_BASE64_BYTES, timeout=60.0, retry_all=True, +async def _materialize_video(video_url: str, task_id: Optional[str], temp_paths: list) -> Path: + """Local video path for a terminal-backend path, local file, or HTTP(S) URL. Only files created + here are appended to ``temp_paths`` — never user-provided paths. Terminal-backend reads use the + shared media resolver with ``permitted=("video",)`` — the exact pipeline vision_analyze uses + (media-cache host reads, bounded in-sandbox exec-read, credential-read guard, 50MB cap).""" + from tools.image_source import ( + ImageResolutionError, ResolveContext, _is_local_terminal_backend, resolve_image_source, ) + source = video_url.removeprefix("file://") + local_path = Path(os.path.expanduser(source)) + lowered = (video_url or "").strip().lower() + path_like = bool(lowered) and not lowered.startswith(("http://", "https://", "data:")) + if not _is_local_terminal_backend() and path_like: + logger.info("Reading video source via terminal backend: %s", video_url) + suffix = Path(source).suffix.lower() + if suffix not in _VIDEO_MIME_TYPES: + raise ValueError(_unsupported_video_format(suffix)) + try: + resolved = await resolve_image_source( + video_url, ResolveContext(task_id=task_id), permitted=("video",)) + except ImageResolutionError as exc: + raise ValueError(f"Could not read video from terminal backend: {exc}") from exc + temp_dir = get_hermes_dir("cache/video", "temp_video_files") + temp_dir.mkdir(parents=True, exist_ok=True) + path = temp_dir / f"terminal_video_{uuid.uuid4()}{suffix}" + path.write_bytes(resolved.data) + temp_paths.append(path) + return path + if local_path.is_file(): + from agent.file_safety import raise_if_read_blocked + raise_if_read_blocked(str(local_path)) + logger.info("Using local video file: %s", video_url) + return local_path + if await _validate_image_url_async(video_url): + blocked = check_website_access(video_url) + if blocked: + raise PermissionError(blocked["message"]) + path = get_hermes_dir("cache/video", "temp_video_files") / f"temp_video_{uuid.uuid4()}.mp4" + temp_paths.append(path) + # Video downloads retry every failure class (legacy behavior). + await _download_media( + video_url, path, 3, media_label="Video", accept="video/*,*/*;q=0.8", + max_bytes=_MAX_VIDEO_BASE64_BYTES, timeout=60.0, retry_all=True) + return path + raise ValueError("Invalid video source. Provide an HTTP/HTTPS URL or a valid local file path.") async def video_analyze_tool( - video_url: str, - user_prompt: str, - model: str = None, - task_id: Optional[str] = None, -) -> str: + video_url: str, user_prompt: str, model: str = None, task_id: Optional[str] = None) -> str: """Analyze a video via multimodal LLM. Returns JSON {success, analysis}.""" - if not isinstance(user_prompt, str): - user_prompt = str(user_prompt) if user_prompt is not None else "" - debug_call_data = _debug_call_data("video", video_url, user_prompt, model) - - temp_video_path = None - should_cleanup = True - - try: - from tools.interrupt import is_interrupted - if is_interrupted(): - return tool_error("Interrupted", success=False) - - logger.info("Analyzing video: %s", video_url[:60]) - logger.info("User prompt: %s", user_prompt[:100]) - - resolved_url = video_url - if resolved_url.startswith("file://"): - resolved_url = resolved_url[len("file://"):] - local_path = Path(os.path.expanduser(resolved_url)) - - from tools.image_source import _is_local_terminal_backend - if not _is_local_terminal_backend() and _is_path_like_video_source(video_url): - logger.info("Reading video source via terminal backend: %s", video_url) - temp_video_path = await _materialize_video_from_terminal_backend(video_url, task_id) - elif local_path.is_file(): - from agent.file_safety import raise_if_read_blocked - raise_if_read_blocked(str(local_path)) - logger.info("Using local video file: %s", video_url) - temp_video_path = local_path - should_cleanup = False - elif await _validate_image_url_async(video_url): - blocked = check_website_access(video_url) - if blocked: - raise PermissionError(blocked["message"]) - temp_dir = get_hermes_dir("cache/video", "temp_video_files") - temp_video_path = temp_dir / f"temp_video_{uuid.uuid4()}.mp4" - await _download_video(video_url, temp_video_path) - else: - raise ValueError( - "Invalid video source. Provide an HTTP/HTTPS URL or a valid local file path." - ) - + async def stage(prompt: str, debug_call_data: dict, temp_paths: list) -> tuple: + temp_video_path = await _materialize_video(video_url, task_id, temp_paths) video_size_bytes = temp_video_path.stat().st_size video_size_mb = video_size_bytes / (1024 * 1024) logger.info("Video ready (%.1f MB)", video_size_mb) - detected_mime = _detect_video_mime_type(temp_video_path) if not detected_mime: raise ValueError(_unsupported_video_format(temp_video_path.suffix)) - if video_size_bytes > _VIDEO_SIZE_WARN_BYTES: logger.warning("Video is %.1f MB — may be slow or rejected", video_size_mb) - video_data_url = _video_to_base64_data_url(temp_video_path, mime_type=detected_mime) if len(video_data_url) > _MAX_VIDEO_BASE64_BYTES: raise ValueError( f"Video too large for API: base64 payload is {len(video_data_url) / (1024 * 1024):.1f} MB " f"(limit {_MAX_VIDEO_BASE64_BYTES / (1024 * 1024):.0f} MB). " - f"Compress or trim the video and retry." - ) - + f"Compress or trim the video and retry.") debug_call_data["video_size_bytes"] = video_size_bytes - - messages = [{ - "role": "user", - "content": [ - {"type": "text", "text": user_prompt}, - {"type": "video_url", "video_url": {"url": video_data_url}}, - ], - }] - - vision_timeout, vision_temperature = _read_vision_call_settings(180.0, min_timeout=180.0) - call_kwargs = { - "task": "vision", - "messages": messages, - "temperature": vision_temperature, - "timeout": vision_timeout, - } - if model: - call_kwargs["model"] = model - + messages = _media_messages(prompt, "video_url", video_data_url) + call_kwargs = _aux_call_kwargs(messages, model, 180.0, min_timeout=180.0) analysis = await _call_vision_llm(call_kwargs, "Empty video response, retrying once") - - analysis_length = len(analysis) if analysis else 0 - logger.info("Video analysis completed (%s characters)", analysis_length) - - result = { - "success": True, - "analysis": analysis or "There was a problem with the request and the video could not be analyzed.", - } - debug_call_data["success"] = True - debug_call_data["analysis_length"] = analysis_length - return _finish_analysis("video_analyze_tool", debug_call_data, result) - - except Exception as e: - error_msg = f"Error analyzing video: {str(e)}" - logger.error("%s", error_msg, exc_info=True) - analysis = _classify_analysis_error( - e, _VIDEO_ERROR_RULES, - "There was a problem with the request and the video could not " - "be analyzed. Error: {e}", - ) - debug_call_data["error"] = error_msg - return _finish_analysis("video_analyze_tool", debug_call_data, { - "success": False, - "error": error_msg, - "analysis": analysis, - }) - - finally: - if should_cleanup: - _cleanup_temp_media(temp_video_path, "video") + return analysis, None + return await _run_analysis("video", video_url, user_prompt, model, stage) VIDEO_ANALYZE_SCHEMA = { @@ -1396,16 +991,12 @@ VIDEO_ANALYZE_SCHEMA = { def _handle_video_analyze(args: Dict[str, Any], **kw: Any) -> Awaitable[str]: - video_url = args.get("video_url", "") - question = args.get("question", "") + video_url, question = args.get("video_url", ""), args.get("question", "") full_prompt = ( "Fully describe and explain everything happening in this video, " "including visual content, motion, audio cues, text overlays, and scene " - f"transitions. Then answer the following question:\n\n{question}" - ) - model = _configured_aux_model( - ("video", "vision"), ("AUXILIARY_VIDEO_MODEL", "AUXILIARY_VISION_MODEL"), - ) + f"transitions. Then answer the following question:\n\n{question}") + model = _configured_aux_model(("video", "vision"), ("AUXILIARY_VIDEO_MODEL", "AUXILIARY_VISION_MODEL")) return video_analyze_tool(video_url, full_prompt, model, task_id=kw.get("task_id")) @@ -1416,5 +1007,4 @@ registry.register( handler=_handle_video_analyze, check_fn=check_vision_requirements, is_async=True, - emoji="🎬", -) + emoji="🎬") diff --git a/tools/vision_tools_image_prep.py b/tools/vision_tools_image_prep.py index 2c67d93e58..908c1ef185 100644 --- a/tools/vision_tools_image_prep.py +++ b/tools/vision_tools_image_prep.py @@ -1,9 +1,8 @@ """Image format detection, normalization and region cropping for vision tools. -Everything here runs BEFORE an image is base64-embedded. A vision tool result -is baked into immutable conversation history and re-sent every turn, so an -unsupported media type or corrupt bytes would wedge the session with a -non-retryable 400 on every resume — normalization must happen up front. +Everything here runs BEFORE an image is base64-embedded: a vision tool result is +baked into immutable history and re-sent every turn, so an unsupported media type +or corrupt bytes would wedge the session with a non-retryable 400 on every resume. """ from __future__ import annotations @@ -30,8 +29,11 @@ _EXTENSION_MIME_TYPES = { # Media types the major vision providers (Anthropic in particular) accept # inline. SVG/BMP/TIFF are rejected with a non-retryable 400. -_ANTHROPIC_SUPPORTED_MEDIA_TYPES = frozenset( - {"image/jpeg", "image/png", "image/gif", "image/webp"} +_ANTHROPIC_SUPPORTED_MEDIA_TYPES = frozenset({"image/jpeg", "image/png", "image/gif", "image/webp"}) + + +_MAGIC_MIME_TYPES = ( + (b"\xff\xd8\xff", "image/jpeg"), ((b"GIF87a", b"GIF89a"), "image/gif"), (b"BM", "image/bmp"), ) @@ -41,17 +43,12 @@ def _determine_mime_type(image_path: Path) -> str: def _detect_image_mime_type_from_bytes(data: bytes) -> Optional[str]: - """Magic-byte MIME sniff (authoritative; no extension trust). - - Returns ``None`` for anything without a recognized header — including SVG, - which has no magic bytes (the resolver sniffs `` Optional[str]: try: with Image.open(BytesIO(data)) as image: image.verify() + return "image/png" except Exception: return None - return "image/png" - if header.startswith(b"\xff\xd8\xff"): - return "image/jpeg" - if header.startswith((b"GIF87a", b"GIF89a")): - return "image/gif" - if header.startswith(b"BM"): - return "image/bmp" + for magic, mime in _MAGIC_MIME_TYPES: + if header.startswith(magic): + return mime if len(header) >= 12 and header[:4] == b"RIFF" and header[8:12] == b"WEBP": return "image/webp" return None def _supported_media_types() -> frozenset: - """Formats the ACTIVE main model's server can decode. - - The managed llama-server decodes with stb_image — no WebP — and an - undecodable image part fails SILENTLY (the model confabulates), so the set - is narrowed there and normalization converts those formats to PNG. - """ + """Formats the ACTIVE main model's server can decode. The managed llama-server decodes with + stb_image — no WebP — and an undecodable image part fails SILENTLY (the model confabulates), + so the set is narrowed there and normalization converts those formats to PNG.""" try: - from agent.auxiliary_client import _runtime_main_value - from hermes_cli.local_runtime.capabilities import ( - ACCEPTED_IMAGE_MIMES, - is_managed_provider, - ) - - if is_managed_provider( - str(_runtime_main_value("provider") or ""), - str(_runtime_main_value("base_url") or "")): + from agent.auxiliary_client import _runtime_main_value as _v + from hermes_cli.local_runtime.capabilities import ACCEPTED_IMAGE_MIMES, is_managed_provider + if is_managed_provider(str(_v("provider") or ""), str(_v("base_url") or "")): return ACCEPTED_IMAGE_MIMES - except Exception: # noqa: BLE001 — best-effort narrowing only + except Exception: # best-effort narrowing only pass return _ANTHROPIC_SUPPORTED_MEDIA_TYPES +def _nonempty_file(path: Path) -> bool: + return path.exists() and path.stat().st_size > 0 + + def _rasterize_svg_to_png(svg_path: Path, out_path: Path) -> bool: """Best-effort SVG → PNG via cairosvg, svglib+reportlab, rsvg-convert, inkscape (all soft deps).""" - def _ok() -> bool: - return out_path.exists() and out_path.stat().st_size > 0 - try: import cairosvg # type: ignore cairosvg.svg2png(url=str(svg_path), write_to=str(out_path)) - return _ok() + return _nonempty_file(out_path) except Exception: pass try: @@ -113,23 +99,18 @@ def _rasterize_svg_to_png(svg_path: Path, out_path: Path) -> bool: drawing = svg2rlg(str(svg_path)) if drawing is not None: renderPM.drawToFile(drawing, str(out_path), fmt="PNG") - return _ok() + return _nonempty_file(out_path) except Exception: pass import shutil import subprocess for cmd in ( ["rsvg-convert", "-o", str(out_path), str(svg_path)], - ["inkscape", str(svg_path), "--export-type=png", - f"--export-filename={out_path}"], - ): + ["inkscape", str(svg_path), "--export-type=png", f"--export-filename={out_path}"]): if shutil.which(cmd[0]): try: - subprocess.run( - cmd, check=True, capture_output=True, timeout=30, - stdin=subprocess.DEVNULL, - ) - if _ok(): + subprocess.run(cmd, check=True, capture_output=True, timeout=30, stdin=subprocess.DEVNULL) + if _nonempty_file(out_path): return True except Exception: continue @@ -137,58 +118,43 @@ def _rasterize_svg_to_png(svg_path: Path, out_path: Path) -> bool: def _normalize_to_supported_image( - image_path: Path, detected_mime: str -) -> tuple[Optional[Path], Optional[str], Optional[str]]: - """Ensure an image is in a provider-supported format. - - Returns ``(path, mime, error)``: the input unchanged when already - supported; ``(new_png_path, "image/png", None)`` after conversion — a temp - file the CALLER must clean up; ``(None, None, message)`` when impossible. - SVG is rasterized; other Pillow-readable rasters (BMP, TIFF) re-encode to PNG. - """ + image_path: Path, detected_mime: str) -> tuple[Optional[Path], Optional[str], Optional[str]]: + """Ensure an image is in a provider-supported format. Returns ``(path, mime, error)``: the input + unchanged when supported; ``(new_png_path, "image/png", None)`` after conversion — a temp file + the CALLER must clean up; ``(None, None, message)`` when impossible. SVG is rasterized; other + Pillow-readable rasters (BMP, TIFF) re-encode to PNG.""" if detected_mime in _supported_media_types(): return image_path, detected_mime, None - out_dir = get_hermes_dir("cache/vision", "temp_vision_images") out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / f"converted_{uuid.uuid4()}.png" - if detected_mime == "image/svg+xml": if _rasterize_svg_to_png(image_path, out_path): return out_path, "image/png", None - return ( - None, - None, + return None, None, ( "This is an SVG, which vision models cannot read directly, and no " "SVG rasterizer is installed (tried cairosvg, svglib, rsvg-convert, " "inkscape). Convert the SVG to PNG first — e.g. open it in a browser " "and screenshot it, or install a rasterizer " - "(`pip install cairosvg`) — then re-run vision_analyze on the PNG.", - ) - + "(`pip install cairosvg`) — then re-run vision_analyze on the PNG.") try: from PIL import Image as _PILImage with _PILImage.open(image_path) as _img: if _img.mode not in ("RGB", "RGBA", "L"): _img = _img.convert("RGBA") _img.save(out_path, format="PNG") - if out_path.exists() and out_path.stat().st_size > 0: + if _nonempty_file(out_path): return out_path, "image/png", None except Exception as _exc: - logger.warning("Failed to normalize %s image to PNG: %s", - detected_mime, _exc) - return ( - None, - None, + logger.warning("Failed to normalize %s image to PNG: %s", detected_mime, _exc) + return None, None, ( f"Image format {detected_mime!r} is not supported by the vision API " f"and could not be converted to PNG (install Pillow for raster " - f"conversion). Convert it to PNG or JPEG and try again.", - ) + f"conversion). Convert it to PNG or JPEG and try again.") -# Full raster validation runs on untrusted images in a shared CPU executor: -# bound animated-image work by frame count AND total decoded area so a compact -# file cannot monopolize a worker with unbounded frames. +# Full raster validation runs on untrusted images in a shared CPU executor: bound animated +# work by frame count AND total decoded area so a compact file cannot monopolize a worker. _VISION_MAX_VALIDATED_FRAME_COUNT = 100 _VISION_MAX_VALIDATED_AGGREGATE_PIXELS = 100_000_000 @@ -196,17 +162,12 @@ _VISION_MAX_VALIDATED_AGGREGATE_PIXELS = 100_000_000 def _validate_raster_image_decodable( image_path: Path, max_frames: int = _VISION_MAX_VALIDATED_FRAME_COUNT, - max_pixels: int = _VISION_MAX_VALIDATED_AGGREGATE_PIXELS, -) -> Optional[str]: - """Return an error unless Pillow can fully decode every frame. - - Header sniffing and ``Image.open`` only inspect containers: a timed-out - download can look like a valid PNG with a truncated pixel stream. Without - Pillow the image passes unvalidated rather than rejecting everything. - """ + max_pixels: int = _VISION_MAX_VALIDATED_AGGREGATE_PIXELS) -> Optional[str]: + """Return an error unless Pillow can fully decode every frame. Header sniffing and ``Image.open`` + only inspect containers: a timed-out download can look like a valid PNG with a truncated pixel + stream. Without Pillow the image passes unvalidated rather than rejecting everything.""" try: - from PIL import Image as _PILImage - from PIL import ImageSequence as _PILImageSequence + from PIL import Image as _PILImage, ImageSequence as _PILImageSequence except ImportError: return None try: @@ -214,23 +175,19 @@ def _validate_raster_image_decodable( image.verify() with _PILImage.open(image_path) as image: validated_pixels = 0 - for frame_number, frame in enumerate( - _PILImageSequence.Iterator(image), start=1 - ): + for frame_number, frame in enumerate(_PILImageSequence.Iterator(image), start=1): if frame_number > max_frames: return ( "Image validation rejected animation: " f"frame {frame_number} exceeds the maximum " - f"{max_frames} validated frames." - ) + f"{max_frames} validated frames.") next_validated_pixels = validated_pixels + frame.width * frame.height if next_validated_pixels > max_pixels: return ( "Image validation rejected animation: aggregate decoded " f"pixel count would reach {next_validated_pixels} at frame " f"{frame_number}, exceeding the maximum " - f"{max_pixels}." - ) + f"{max_pixels}.") frame.load() validated_pixels = next_validated_pixels except Exception as exc: @@ -239,12 +196,9 @@ def _validate_raster_image_decodable( def _image_exceeds_dimension(image_path: Path, max_dimension: int) -> bool: - """True if the longest side exceeds ``max_dimension`` px. - - Anthropic enforces an 8000px per-side cap independently of the byte cap. - Returns False (no forced resize) without Pillow or on unreadable files — - a missing soft dependency must never break the embed path. - """ + """True if the longest side exceeds ``max_dimension`` px (Anthropic's 8000px per-side cap is + independent of bytes). False without Pillow or on unreadable files — a missing soft + dependency must never break the embed path.""" try: from PIL import Image as _PILImage with _PILImage.open(image_path) as _img: @@ -254,57 +208,39 @@ def _image_exceeds_dimension(image_path: Path, max_dimension: int) -> bool: def _crop_image_region( - image_path: Path, - region: Any, - offset_out: Optional[dict] = None, + image_path: Path, region: Any, offset_out: Optional[dict] = None ) -> tuple[Optional[Path], Optional[str], Optional[str]]: - """Crop to ``region`` = [x1, y1, x2, y2] (original-image pixels). - - Applied BEFORE downscaling so the crop gets the full resolution budget. - Coordinates clamp to the image bounds; a zero-area/inverted region is - rejected with an error naming the real dimensions. Returns - ``(cropped_temp_path, mime, None)`` — caller owns cleanup — or - ``(None, None, error)``. Ported from QwenLM/qwen-code zoom-image.ts (Apache-2.0). - """ + """Crop to ``region`` = [x1, y1, x2, y2] (original-image pixels), BEFORE downscaling so the crop + gets the full resolution budget. Coordinates clamp to the image bounds; a zero-area/inverted + region is rejected with an error naming the real dimensions. Returns ``(cropped_temp_path, + mime, None)`` — caller owns cleanup — or ``(None, None, error)``. + Ported from QwenLM/qwen-code zoom-image.ts (Apache-2.0).""" try: from PIL import Image except ImportError: return None, None, ( "region cropping requires Pillow (`pip install Pillow`); " - "retry without the region parameter." - ) - - if ( - not isinstance(region, (list, tuple)) - or len(region) != 4 - or not all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in region) - ): + "retry without the region parameter.") + if not (isinstance(region, (list, tuple)) and len(region) == 4 + and all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in region)): return None, None, ( "Invalid region: expected [x1, y1, x2, y2] as four numbers " - "(pixel coordinates in the original image)." - ) - + "(pixel coordinates in the original image).") try: with Image.open(image_path) as img: width, height = img.size x1, y1, x2, y2 = (int(v) for v in region) - cx1 = max(0, min(x1, width)) - cy1 = max(0, min(y1, height)) - cx2 = max(0, min(x2, width)) - cy2 = max(0, min(y2, height)) + cx1, cy1, cx2, cy2 = (max(0, min(v, b)) for v, b in zip((x1, y1, x2, y2), (width, height) * 2)) if cx2 <= cx1 or cy2 <= cy1: return None, None, ( f"Invalid region [{x1}, {y1}, {x2}, {y2}]: crops to zero " f"area after clamping to the image bounds. The image is " f"{width}x{height} px — pick x1