diff --git a/agent/delegation_context.py b/agent/delegation_context.py index 5ff3eb8e8c..e1aa306d1b 100644 --- a/agent/delegation_context.py +++ b/agent/delegation_context.py @@ -1,9 +1,9 @@ """Context-local state for delegate_task child execution. -A Hermes process may itself be a Kanban dispatcher worker with HERMES_KANBAN_* -in os.environ. delegate_task children (in-process) and cron jobs fired via -``cronjob(action="run")`` are NOT dispatcher-owned, so identity gates must -fail closed for them without mutating the process-global environment. +A Hermes process may itself be a Kanban dispatcher worker with HERMES_KANBAN_* in +os.environ. In-process delegate_task children and cron jobs fired via +``cronjob(action="run")`` are NOT dispatcher-owned, so identity gates must fail +closed for them without mutating the process-global environment. """ from __future__ import annotations @@ -13,36 +13,26 @@ from contextvars import ContextVar, Token from typing import Iterator, Mapping, MutableMapping _DELEGATED_CHILD_CONTEXT: ContextVar[bool] = ContextVar("hermes_delegated_child_context", default=False) - -# Any in-process execution that is NOT the dispatcher-owned worker (cron jobs). -# Kept separate from _DELEGATED_CHILD_CONTEXT so delegate_task-specific -# behaviour (subprocess env scrubbing, its error strings) is unchanged. +# Any in-process execution that is NOT the dispatcher-owned worker (cron jobs). Kept separate +# so delegate_task-specific behaviour (subprocess env scrubbing, its error strings) is unchanged. _NON_DISPATCHER_OWNED_CONTEXT: ContextVar[bool] = ContextVar("hermes_non_dispatcher_owned_context", default=False) DELEGATED_CHILD_ENV_MARKER = "HERMES_DELEGATED_CHILD_CONTEXT" KANBAN_ENV_KEYS: tuple[str, ...] = ( - "HERMES_KANBAN_TASK", - "HERMES_KANBAN_RUN_ID", - "HERMES_KANBAN_WORKSPACE", - "HERMES_KANBAN_WORKSPACES_ROOT", - "HERMES_KANBAN_CLAIM_LOCK", - "HERMES_KANBAN_BOARD", - "HERMES_KANBAN_DB", + "HERMES_KANBAN_TASK", "HERMES_KANBAN_RUN_ID", "HERMES_KANBAN_WORKSPACE", "HERMES_KANBAN_WORKSPACES_ROOT", + "HERMES_KANBAN_CLAIM_LOCK", "HERMES_KANBAN_BOARD", "HERMES_KANBAN_DB", ) @contextmanager def delegated_child_context(session_id: str | None = None) -> Iterator[None]: - """Mark child execution and isolate its task-local session identity. - - Even a context entered without an id must restore the parent's session - ContextVar (child construction calls ``set_current_session_id``). - """ + """Mark child execution and isolate its task-local session identity. Even a context + entered without an id must restore the parent's session ContextVar (child + construction calls ``set_current_session_id``).""" token = _DELEGATED_CHILD_CONTEXT.set(True) try: - # Lazy: session_context calls is_delegated_child_context(). - from gateway.session_context import scoped_current_session_id + from gateway.session_context import scoped_current_session_id # lazy: it calls is_delegated_child_context() with scoped_current_session_id(session_id): yield @@ -67,13 +57,10 @@ def exit_non_dispatcher_owned_context(token: Token[bool]) -> None: @contextmanager def non_dispatcher_owned_context() -> Iterator[None]: - """Mark in-process execution that does NOT own the dispatcher's Kanban task. - - Without it a cron agent run inside a worker is misread as that worker - (kanban toolset force-added, ``kanban_complete`` defaulting to its task). - ContextVar-scoped rather than clearing os.environ, which the worker's claim - heartbeat and concurrent readers share. - """ + """Mark in-process execution that does NOT own the dispatcher's Kanban task; without it + a cron agent run inside a worker is misread as that worker (kanban toolset force-added, + ``kanban_complete`` defaulting to its task). ContextVar-scoped rather than clearing + os.environ, which the worker's claim heartbeat and concurrent readers share.""" token = enter_non_dispatcher_owned_context() try: yield @@ -92,10 +79,8 @@ def is_delegated_child_process_context() -> bool: def scrub_kanban_env(env: Mapping[str, str] | MutableMapping[str, str]) -> dict[str, str]: - """Return *env* with dispatcher-only Kanban variables removed.""" - cleaned = dict(env) - for key in KANBAN_ENV_KEYS: - cleaned.pop(key, None) + """Return *env* with dispatcher-only Kanban variables removed and the lineage marker set.""" + cleaned = {k: v for k, v in env.items() if k not in KANBAN_ENV_KEYS} cleaned[DELEGATED_CHILD_ENV_MARKER] = "1" return cleaned @@ -103,11 +88,8 @@ def scrub_kanban_env(env: Mapping[str, str] | MutableMapping[str, str]) -> dict[ def delegated_child_subprocess_env( env: Mapping[str, str] | MutableMapping[str, str] | None = None, ) -> dict[str, str] | None: - """Return an env override only when delegated-child lineage must cross fork. - - Preserves ``env=None`` inherit semantics for non-delegated calls; in a - child, materializes a scrubbed env carrying the lineage marker. - """ + """Env override only when delegated-child lineage must cross fork: preserves ``env=None`` + inherit semantics for non-delegated calls; in a child, a scrubbed env carrying the marker.""" if not is_delegated_child_process_context(): return None if env is None else dict(env) return scrub_kanban_env(os.environ if env is None else env) diff --git a/agent/estop.py b/agent/estop.py index 59cb0ffbc4..86c5d585ee 100644 --- a/agent/estop.py +++ b/agent/estop.py @@ -1,13 +1,11 @@ """Global emergency stop (ESTOP) — a resumable pause for NEW work only. ``hermes pause`` writes a sentinel at ``$HERMES_HOME/ESTOP``; ``hermes resume`` -removes it. While it exists the cron scheduler, kanban dispatcher and new -gateway turns skip work; in-flight work is never killed. The check is one or -two ``os.stat`` calls (process home + fleet root when they differ), uncached, -so engaging/disengaging binds on the next check. The sentinel body is optional +removes it. While it exists the cron scheduler, kanban dispatcher and new gateway +turns skip work; in-flight work is never killed. The check is one or two uncached +``os.stat`` calls (process home + fleet root when they differ). The body is optional JSON ``{"reason", "engaged_at"}``; a corrupt/empty file still counts as engaged -(fail safe, e.g. ``touch ~/.hermes/ESTOP``). Ported from gastownhall/gastown -estop.go (MIT). +(fail safe, e.g. ``touch ~/.hermes/ESTOP``). Ported from gastownhall/gastown estop.go (MIT). """ from __future__ import annotations @@ -15,6 +13,7 @@ from __future__ import annotations import json import logging import threading +from contextlib import suppress from datetime import datetime, timezone from pathlib import Path from typing import Optional @@ -24,8 +23,7 @@ from agent.file_safety import _hermes_home_path as _hermes_home, _hermes_root_pa SENTINEL_NAME = "ESTOP" -# Per-component "logged already for this engagement" flags: log once per -# engagement, not once per tick. +# Per-component "logged already for this engagement" flags: log once per engagement, not per tick. _log_lock = threading.Lock() _logged_components: set[str] = set() @@ -39,19 +37,16 @@ def _candidate_sentinel_paths() -> list: """Profile home first, then the fleet root if it is a different directory: a profile gateway (HERMES_HOME=~/.hermes/profiles/) must still honor an operator's ~/.hermes/ESTOP.""" primary = sentinel_path() - paths = [primary] try: root = _canonical_root() / SENTINEL_NAME except Exception: - return paths + return [primary] try: - if root.resolve() != primary.resolve(): - paths.append(root) + distinct = root.resolve() != primary.resolve() except Exception: # Non-Path test doubles fail .resolve(); plain equality still dedupes. - if root != primary: - paths.append(root) - return paths + distinct = root != primary + return [primary, root] if distinct else [primary] def is_engaged() -> bool: @@ -74,11 +69,8 @@ def engage(reason: Optional[str] = None) -> Path: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") except OSError: - # Best effort: an empty/partial sentinel still pauses (fail safe). - try: + with suppress(OSError): # Best effort: an empty/partial sentinel still pauses (fail safe). path.touch(exist_ok=True) - except OSError: - pass return path @@ -95,35 +87,27 @@ def disengage() -> bool: def get_state() -> Optional[dict]: - """Return ``{"reason", "engaged_at"}`` or None when not engaged. - - An unreadable/corrupt body still reports engaged with both fields None. - """ + """Return ``{"reason", "engaged_at"}`` or None when not engaged; an unreadable/corrupt + body still reports engaged with both fields None.""" if not is_engaged(): return None - reason = engaged_at = None + state = {"reason": None, "engaged_at": None} found = False for path in _candidate_sentinel_paths(): try: - exists = path.exists() + if not path.exists(): + continue except OSError: - return {"reason": None, "engaged_at": None} + return state except AttributeError: continue - if not exists: - continue found = True - try: + with suppress(OSError, ValueError, AttributeError): raw = json.loads(path.read_text(encoding="utf-8")) if isinstance(raw, dict): - reason = raw.get("reason") or None - engaged_at = raw.get("engaged_at") or None + state = {"reason": raw.get("reason") or None, "engaged_at": raw.get("engaged_at") or None} break - except (OSError, ValueError, AttributeError): - continue - if not found: - return None - return {"reason": reason, "engaged_at": engaged_at} + return state if found else None def paused_reply() -> Optional[str]: @@ -131,31 +115,24 @@ def paused_reply() -> Optional[str]: state = get_state() if state is None: return None - reason = state.get("reason") - tag = f" ({reason})" if reason else "" + tag = f" ({state['reason']})" if state.get("reason") else "" return f"⏸️ Hermes is paused{tag}. New work is on hold; run `hermes resume` to pick things back up." def check_paused(component: str, logger: logging.Logger) -> bool: - """Return True when engaged, logging once per engagement per component - (re-armed after a resume).""" + """Return True when engaged, logging once per engagement per component (re-armed after a resume).""" if not is_engaged(): with _log_lock: _logged_components.discard(component) return False with _log_lock: first = component not in _logged_components - if first: - _logged_components.add(component) + _logged_components.add(component) if first: - state = get_state() or {} - reason = state.get("reason") + reason = (get_state() or {}).get("reason") suffix = f" (reason: {reason})" if reason else "" logger.info( - "%s dispatch paused by global emergency stop%s — remove with " - "`hermes resume` (%s)", - component, - suffix, - sentinel_path(), + "%s dispatch paused by global emergency stop%s — remove with `hermes resume` (%s)", + component, suffix, sentinel_path(), ) return True diff --git a/agent/file_safety.py b/agent/file_safety.py index c878c087d4..d3a981d090 100644 --- a/agent/file_safety.py +++ b/agent/file_safety.py @@ -9,6 +9,7 @@ from __future__ import annotations import os from pathlib import Path +from contextlib import suppress from typing import Optional @@ -38,101 +39,79 @@ def _hermes_dirs() -> list[Path]: Both are checked so credential stores at /... stay guarded when running under a profile (HERMES_HOME = /profiles/). """ - dirs: list[Path] = [] - for base in (_hermes_home_path(), _hermes_root_path()): + return list(dict.fromkeys(_resolve_each((_hermes_home_path(), _hermes_root_path())))) + + +def _resolve_each(paths) -> list[Path]: + """``p.resolve()`` for each path, skipping ones that fail to resolve.""" + out: list[Path] = [] + for p in paths: + with suppress(Exception): + out.append(p.resolve()) + return out + + +def _is_under(resolved: str | Path, base: str | Path) -> bool: + """True when ``resolved`` equals ``base`` or lies below it (both already resolved); + ``Path`` inputs use ``relative_to`` (platform semantics), ``str`` a realpath prefix test.""" + if isinstance(resolved, Path): try: - real = base.resolve() - except Exception: - continue - if real not in dirs: - dirs.append(real) - return dirs + return resolved.relative_to(base) is not None + except ValueError: + return False + return resolved == base or resolved.startswith(str(base) + os.sep) -def _is_under(resolved: str, prefix: str) -> bool: - return resolved == prefix or resolved.startswith(prefix + os.sep) - - -def _under_any(resolved: Path, base: Path) -> bool: - try: - resolved.relative_to(base) - return True - except ValueError: - return False +def _resolve_target(path: str) -> Optional[Path]: + """``Path(expanduser(path)).resolve()``, or None when resolution fails.""" + with suppress(OSError, RuntimeError): + return Path(os.path.expanduser(str(path))).resolve() + return None def _home_and_resolved(path: str) -> tuple[str, str]: """``(realpath(~), realpath(expanduser(path)))`` — the write-guard coordinate pair.""" - return os.path.realpath(os.path.expanduser("~")), os.path.realpath(os.path.expanduser(str(path))) + return tuple(os.path.realpath(os.path.expanduser(p)) for p in ("~", str(path))) def build_write_denied_paths(home: str) -> set[str]: """Return exact sensitive paths that must never be written.""" - hermes_home = _hermes_home_path() - hermes_root = _hermes_root_path() - return { - os.path.realpath(p) - for p in [ - os.path.join(home, ".ssh", "authorized_keys"), - os.path.join(home, ".ssh", "id_rsa"), - os.path.join(home, ".ssh", "id_ed25519"), - # ``~/.ssh/config`` is deliberately NOT hard-denied: no key bytes, and - # editing it is routine. It can carry ProxyCommand / Match exec, so it - # goes through the approval gate instead (build_write_approval_paths). - # Both the active-profile and top-level .env: overwriting the root .env - # leaks credentials across every profile that inherits from it. - str(hermes_home / ".env"), - str(hermes_root / ".env"), - # Anthropic PKCE credential stores; the root copy is still read by - # default/non-profile sessions when a profile is active. - str(hermes_home / ".anthropic_oauth.json"), - str(hermes_root / ".anthropic_oauth.json"), - # Bitwarden Secrets Manager encrypted disk cache. - str(hermes_home / "cache" / "bws_cache.enc.json"), - str(hermes_root / "cache" / "bws_cache.enc.json"), - os.path.join(home, ".netrc"), - os.path.join(home, ".pgpass"), - os.path.join(home, ".npmrc"), - os.path.join(home, ".pypirc"), - os.path.join(home, ".git-credentials"), - "/etc/sudoers", - "/etc/passwd", - "/etc/shadow", - ] - } + # ``~/.ssh/config`` is deliberately NOT hard-denied: no key bytes, and editing + # it is routine. It can carry ProxyCommand / Match exec, so it goes through the + # approval gate instead (build_write_approval_paths). + home_files = ( + (".ssh", "authorized_keys"), (".ssh", "id_rsa"), (".ssh", "id_ed25519"), + (".netrc",), (".pgpass",), (".npmrc",), (".pypirc",), (".git-credentials",), + ) + # Both the active-profile and top-level copies: overwriting the root .env leaks + # credentials across every profile that inherits from it; the root Anthropic + # PKCE store is still read by default/non-profile sessions when a profile is + # active; bws_cache.enc.json is the Bitwarden Secrets Manager encrypted cache. + hermes_files = (".env", ".anthropic_oauth.json", os.path.join("cache", "bws_cache.enc.json")) + paths = [ + *(os.path.join(home, *f) for f in home_files), + *(str(base / f) for f in hermes_files for base in (_hermes_home_path(), _hermes_root_path())), + "/etc/sudoers", "/etc/passwd", "/etc/shadow", + ] + return {os.path.realpath(p) for p in paths} def build_write_denied_prefixes(home: str) -> list[str]: """Return sensitive directory prefixes that must never be written.""" - return [ - os.path.realpath(p) + os.sep - for p in [ - os.path.join(home, ".ssh"), - os.path.join(home, ".aws"), - os.path.join(home, ".gnupg"), - os.path.join(home, ".kube"), - "/etc/sudoers.d", - "/etc/systemd", - os.path.join(home, ".docker"), - os.path.join(home, ".azure"), - os.path.join(home, ".config", "gh"), - os.path.join(home, ".config", "gcloud"), - ] + paths = [ + *(os.path.join(home, d) for d in (".ssh", ".aws", ".gnupg", ".kube")), + "/etc/sudoers.d", "/etc/systemd", + *(os.path.join(home, *d) for d in ((".docker",), (".azure",), (".config", "gh"), (".config", "gcloud"))), ] + return [os.path.realpath(p) + os.sep for p in paths] def get_safe_write_roots() -> set[str]: """Resolved HERMES_WRITE_SAFE_ROOT paths (``os.pathsep``-separated list).""" - env = os.getenv("HERMES_WRITE_SAFE_ROOT", "") - if not env: - return set() roots: set[str] = set() - for path in env.split(os.pathsep): - if path: - try: - roots.add(os.path.realpath(os.path.expanduser(path))) - except (OSError, ValueError): - continue + for path in filter(None, os.getenv("HERMES_WRITE_SAFE_ROOT", "").split(os.pathsep)): + with suppress(OSError, ValueError): + roots.add(os.path.realpath(os.path.expanduser(path))) return roots @@ -170,11 +149,9 @@ def _classify_write_denial(path: str) -> Optional[str]: for base in _hermes_dirs(): for sub in _HERMES_PROTECTED_SUBPATHS: - try: + with suppress(Exception): if _is_under(resolved, os.path.realpath(os.path.join(str(base), sub))): return "credential" - except Exception: - pass safe_roots = get_safe_write_roots() if safe_roots and not any(_is_under(resolved, root) for root in safe_roots): @@ -191,15 +168,13 @@ def is_write_denied(path: str) -> bool: def get_write_denied_error(path: str, *, verb: str = "Write") -> Optional[str]: """Return a user/model-facing error when writes to ``path`` are blocked.""" denial = _classify_write_denial(path) - if denial is None: - return None if denial == "safe_root": roots_display = os.pathsep.join(sorted(get_safe_write_roots())) return ( f"{verb} denied: '{path}' is outside HERMES_WRITE_SAFE_ROOT " f"({roots_display}). Unset the variable or add this path's directory prefix." ) - return f"{verb} denied: '{path}' is a protected system/credential file." + return f"{verb} denied: '{path}' is a protected system/credential file." if denial else None def is_write_approval_required(path: str) -> bool: @@ -211,13 +186,7 @@ def is_write_approval_required(path: str) -> bool: # Secret-bearing project-local env file basenames, blocked anywhere on disk. _BLOCKED_PROJECT_ENV_BASENAMES: set[str] = { - ".env", - ".env.local", - ".env.development", - ".env.production", - ".env.test", - ".env.staging", - ".envrc", + ".env", ".env.local", ".env.development", ".env.production", ".env.test", ".env.staging", ".envrc", } _DID_SUFFIX = ( @@ -226,33 +195,22 @@ _DID_SUFFIX = ( # Exact-file credential stores under HERMES_HOME / . The agent never # needs these directly — provider tools consume them through internal channels. +# bws_cache.json is the Bitwarden Secrets Manager disk cache: plaintext secret values. _CREDENTIAL_FILE_NAMES = ( - "auth.json", - "auth.lock", - ".anthropic_oauth.json", - ".env", - "webhook_subscriptions.json", - os.path.join("auth", "google_oauth.json"), - # Bitwarden Secrets Manager disk cache: plaintext secret values. - os.path.join("cache", "bws_cache.json"), + "auth.json", "auth.lock", ".anthropic_oauth.json", ".env", "webhook_subscriptions.json", + os.path.join("auth", "google_oauth.json"), os.path.join("cache", "bws_cache.json"), ) # Directory-prefix read denies under HERMES_HOME / : (subdir, message for # the directory itself, message for a file inside). browser-profile/ is a copy # of the user's Cookies / Login Data — the same credential class as auth.json. _READ_DENIED_DIRS = ( - ( - "mcp-tokens", - "is the Hermes MCP token directory and cannot be read directly.", - "is a Hermes MCP token file and cannot be read directly.", - ), - ( - "browser-profile", - "is the Hermes real-profile browser snapshot directory (copied " - "cookies/logins) and cannot be read directly.", - "is inside the Hermes real-profile browser snapshot (copied " - "cookies/logins) and cannot be read directly.", - ), + ("mcp-tokens", + "is the Hermes MCP token directory and cannot be read directly.", + "is a Hermes MCP token file and cannot be read directly."), + ("browser-profile", + "is the Hermes real-profile browser snapshot directory (copied cookies/logins) and cannot be read directly.", + "is inside the Hermes real-profile browser snapshot (copied cookies/logins) and cannot be read directly."), ) @@ -270,47 +228,31 @@ def get_read_block_error(path: str) -> Optional[str]: """ resolved = Path(path).expanduser().resolve() hermes_dirs = _hermes_dirs() - - for hd in hermes_dirs: - if _under_any(resolved, hd / "skills" / ".hub"): - return ( - f"Access denied: {path} is an internal Hermes cache file " - "and cannot be read directly to prevent prompt injection. " - "Use the skills_list or skill_view tools instead." - ) - - for hd in hermes_dirs: - for name in _CREDENTIAL_FILE_NAMES: - try: - blocked = (hd / name).resolve() - except Exception: - continue - if resolved == blocked: - return ( - f"Access denied: {path} is a Hermes credential store " - "and cannot be read directly. Provider tools consume " - "these credentials through internal channels." + _DID_SUFFIX - ) - - for subdir, dir_msg, file_msg in _READ_DENIED_DIRS: - for hd in hermes_dirs: - try: - blocked_dir = (hd / subdir).resolve() - except Exception: - continue - if resolved == blocked_dir: - return f"Access denied: {path} {dir_msg}{_DID_SUFFIX}" - if _under_any(resolved, blocked_dir): - return f"Access denied: {path} {file_msg}{_DID_SUFFIX}" - - if resolved.name.lower() in _BLOCKED_PROJECT_ENV_BASENAMES: - return ( - f"Access denied: {path} is a secret-bearing environment file " - "and cannot be read to prevent credential leakage. " - "If you need to check the file structure, read .env.example instead." + _DID_SUFFIX + reason = None + if any(_is_under(resolved, hd / "skills" / ".hub") for hd in hermes_dirs): + reason = ( + "is an internal Hermes cache file and cannot be read directly to prevent " + "prompt injection. Use the skills_list or skill_view tools instead." ) - - return None + elif any(resolved in _resolve_each(hd / name for hd in hermes_dirs) for name in _CREDENTIAL_FILE_NAMES): + reason = ( + "is a Hermes credential store and cannot be read directly. Provider tools " + "consume these credentials through internal channels." + _DID_SUFFIX + ) + else: + for subdir, dir_msg, file_msg in _READ_DENIED_DIRS: + for blocked_dir in _resolve_each(hd / subdir for hd in hermes_dirs): + if _is_under(resolved, blocked_dir): + reason = (dir_msg if resolved == blocked_dir else file_msg) + _DID_SUFFIX + break + if reason: + break + if reason is None and resolved.name.lower() in _BLOCKED_PROJECT_ENV_BASENAMES: + reason = ( + "is a secret-bearing environment file and cannot be read to prevent credential " + "leakage. If you need to check the file structure, read .env.example instead." + _DID_SUFFIX + ) + return f"Access denied: {path} {reason}" if reason else None def raise_if_read_blocked(path: str) -> None: @@ -332,33 +274,18 @@ def _resolve_active_profile_name() -> str: """Active profile name from HERMES_HOME: ``~/.hermes`` -> ``"default"``, ``~/.hermes/profiles/X`` -> ``"X"``; ``"default"`` on any resolution failure.""" try: - home_real = _hermes_home_path().resolve() - root_real = _hermes_root_path().resolve() - except (OSError, RuntimeError): + parts = _hermes_home_path().resolve().relative_to(_hermes_root_path().resolve() / "profiles").parts + except (OSError, RuntimeError, ValueError): return "default" - try: - parts = home_real.relative_to(root_real / "profiles").parts - if parts: - return parts[0] - except ValueError: - pass - return "default" - - -def get_cross_profile_warning(path: str) -> Optional[str]: - """RETIRED: always ``None``. Profiles were never isolated (same OS user), so - the guard was ceremony that taught a bypass arg. Stub kept so external - callers/plugins fail soft; the system prompt's active-profile hint remains.""" - return None + return parts[0] if parts else "default" # --- Sandbox-mirror write guard --- # Non-local terminal backends bind a sandbox-local dir to the container's $HOME: # /profiles//sandboxes///home/.hermes/... -# A host-side write there lands on a mirror the host never reads: silent -# success, divergent copies. Path-shape-only detection, independent of the -# active profile. Does NOT cover the inner-container case where the bind mount -# strips the prefix — that is classify_container_mirror_target below. +# A host-side write there lands on a mirror the host never reads: silent success, +# divergent copies. Path-shape-only detection, independent of the active profile; +# the inner-container case (bind mount strips the prefix) is classify_container_mirror_target. _SANDBOX_MIRROR_WARNING = ( "Sandbox-mirror write blocked by soft guard: {target_path} " @@ -369,50 +296,39 @@ _SANDBOX_MIRROR_WARNING = ( ) +def _mirror_info(target: Path, mirror_root: Path, inner_path: str) -> dict: + """Common ``classify_*_mirror_target`` result shape.""" + return {"target_path": str(target), "mirror_root": str(mirror_root), "inner_path": inner_path} + + def classify_sandbox_mirror_target(path: str) -> Optional[dict]: - """Classify a write target as a sandbox-mirror of authoritative Hermes state. - - Returns ``None`` for non-mirror paths, else ``target_path`` (resolved), - ``mirror_root`` (the ``…/home/.hermes`` prefix) and ``inner_path`` (what - the agent likely meant to address on the host). - """ - try: - target = Path(os.path.expanduser(str(path))).resolve() - except (OSError, RuntimeError): - return None - - parts = target.parts + """Classify a write target as a sandbox-mirror of authoritative Hermes state: ``None`` + for non-mirror paths, else ``target_path`` (resolved), ``mirror_root`` (the + ``…/home/.hermes`` prefix) and ``inner_path`` (what the agent meant on the host).""" + target = _resolve_target(path) + parts = target.parts if target is not None else () # Need at least: sandboxes / / / home / .hermes / ; inner_idx = the .hermes part. - for i, part in enumerate(parts): - if part == "sandboxes" and i + 5 < len(parts) and parts[i + 3] == "home" and parts[i + 4] == ".hermes": - inner_idx = i + 4 - break - else: + inner_idx = next( + (i + 4 for i, part in enumerate(parts) + if part == "sandboxes" and i + 5 < len(parts) and parts[i + 3] == "home" and parts[i + 4] == ".hermes"), + None, + ) + if inner_idx is None: return None - return { - "target_path": str(target), - "mirror_root": str(Path(*parts[: inner_idx + 1])), - "inner_path": str(Path(*parts[inner_idx + 1 :])) if inner_idx + 1 < len(parts) else "", - } + inner = str(Path(*parts[inner_idx + 1:])) if inner_idx + 1 < len(parts) else "" + return _mirror_info(target, Path(*parts[: inner_idx + 1]), inner) def _mirror_warning(info: Optional[dict], body: str, bypass: str) -> Optional[str]: """Render ``_SANDBOX_MIRROR_WARNING`` for a classify_* result (``body`` may use ``{inner_path}``).""" if info is None: return None - return _SANDBOX_MIRROR_WARNING.format( - target_path=info["target_path"], - mirror_root=info["mirror_root"], - body=body.format(inner_path=info["inner_path"]), - bypass=bypass, - ) + return _SANDBOX_MIRROR_WARNING.format(**info, body=body.format(inner_path=info["inner_path"]), bypass=bypass) def get_sandbox_mirror_warning(path: str) -> Optional[str]: - """Model-facing soft-guard warning when ``path`` lands in a sandbox mirror, else ``None``. - - Caller surfaces it as a tool-result error; ``cross_profile=True`` bypasses. - """ + """Model-facing soft-guard warning when ``path`` lands in a sandbox mirror, else ``None``; + the caller surfaces it as a tool-result error and ``cross_profile=True`` bypasses.""" return _mirror_warning( classify_sandbox_mirror_target(path), "a per-task mirror created by a non-local terminal backend (docker/daytona/etc.). " @@ -422,37 +338,18 @@ def get_sandbox_mirror_warning(path: str) -> Optional[str]: ) -def classify_container_mirror_target( - path: str, - mirror_prefix: str | None = None, -) -> Optional[dict]: - """Classify a write target as a container-side sandbox mirror. - - Inside the container the bind mount strips the ``sandboxes/`` prefix (the - agent sees plain ``/root/.hermes/…``), so the caller must supply - ``mirror_prefix`` once it knows file tools run in a docker sandbox. - Returns ``None`` without a prefix or when the path is outside it, else - ``target_path``, ``mirror_root`` and ``inner_path``. - """ - if not mirror_prefix: +def classify_container_mirror_target(path: str, mirror_prefix: str | None = None) -> Optional[dict]: + """Classify a write target as a container-side sandbox mirror. Inside the container + the bind mount strips the ``sandboxes/`` prefix (the agent sees plain ``/root/.hermes/…``), + so the caller supplies ``mirror_prefix`` once it knows file tools run in a docker sandbox. + ``None`` without a prefix or outside it, else ``target_path``/``mirror_root``/``inner_path``.""" + target, mirror = _resolve_target(path), _resolve_target(mirror_prefix) if mirror_prefix else None + if target is None or mirror is None or not _is_under(target, mirror): return None - try: - target = Path(os.path.expanduser(str(path))).resolve() - mirror = Path(os.path.expanduser(mirror_prefix)).resolve() - inner = target.relative_to(mirror) - except (OSError, RuntimeError, ValueError): - return None - return { - "target_path": str(target), - "mirror_root": str(mirror), - "inner_path": inner.as_posix(), - } + return _mirror_info(target, mirror, target.relative_to(mirror).as_posix()) -def get_container_mirror_warning( - path: str, - mirror_prefix: str | None = None, -) -> Optional[str]: +def get_container_mirror_warning(path: str, mirror_prefix: str | None = None) -> Optional[str]: """Model-facing soft-guard warning when ``path`` lands in the container's mirror, else ``None``.""" return _mirror_warning( classify_container_mirror_target(path, mirror_prefix), diff --git a/agent/image_gen_provider.py b/agent/image_gen_provider.py index 86857eed51..250c825ad0 100644 --- a/agent/image_gen_provider.py +++ b/agent/image_gen_provider.py @@ -1,29 +1,13 @@ -""" -Image Generation Provider ABC -============================= +"""Image generation provider ABC. -Pluggable-backend interface for image generation. Providers register via -``PluginContext.register_image_gen_provider()``; the one selected by -``image_gen.provider`` services every ``image_generate`` call. Providers live in -``/plugins/image_gen//`` (built-in) or -``~/.hermes/plugins/image_gen//`` (user, opt-in). - -One tool covers text-to-image and image-to-image/editing: the presence of -``image_url`` (and/or ``reference_image_urls``) routes to the provider's edit -endpoint, otherwise text-to-image. Users pick one model; the provider picks the -endpoint. Mirrors ``agent/video_gen_provider.py`` so the two stay learnable. - -Response shape (built by :func:`success_response` / :func:`error_response`):: - - success bool - image str | None URL or absolute file path - model str provider-specific model identifier - prompt str echoed prompt - aspect_ratio str "landscape" | "square" | "portrait" - modality str "text" | "image" (which mode was used) - provider str provider name (for diagnostics) - error str only when success=False - error_type str only when success=False +Providers register via ``PluginContext.register_image_gen_provider()`` (from +``/plugins/image_gen//`` or ``~/.hermes/plugins/image_gen//``); +the one selected by ``image_gen.provider`` services every ``image_generate`` call. +One tool covers text-to-image and editing: ``image_url`` / ``reference_image_urls`` +route to the provider's edit endpoint, otherwise text-to-image. Mirrors +``agent/video_gen_provider.py``. Response dicts come from :func:`success_response` +/ :func:`error_response` (``success, image, model, prompt, aspect_ratio, modality, +provider`` + ``error, error_type`` on failure). """ from __future__ import annotations @@ -44,25 +28,15 @@ DEFAULT_ASPECT_RATIO = "landscape" class ImageGenProvider(CatalogProviderBase): - """Abstract base class for an image generation backend. - - Subclasses must implement :attr:`name` and :meth:`generate`; everything else - has defaults. ``list_models`` entries may add ``speed`` / ``strengths`` / - ``price`` for the picker. - """ + """Abstract base class for an image generation backend. Subclasses implement + :attr:`name` and :meth:`generate`; ``list_models`` entries may add + ``speed`` / ``strengths`` / ``price`` for the picker.""" def capabilities(self) -> Dict[str, Any]: - """What this provider supports: ``modalities`` (``"text"`` and/or - ``"image"``) and ``max_reference_images``. - - The tool layer surfaces this in the dynamic schema so the model knows - when ``image_url`` is honored. Default is text-only so a provider that - doesn't override advertises only text-to-image (backward compatible). - """ - return { - "modalities": ["text"], - "max_reference_images": 0, - } + """``modalities`` (``"text"`` and/or ``"image"``) and ``max_reference_images``. + Surfaced in the dynamic tool schema so the model knows when ``image_url`` is + honored; the text-only default keeps non-overriding providers backward compatible.""" + return {"modalities": ["text"], "max_reference_images": 0} @abc.abstractmethod def generate( @@ -74,86 +48,45 @@ class ImageGenProvider(CatalogProviderBase): reference_image_urls: Optional[List[str]] = None, **kwargs: Any, ) -> Dict[str, Any]: - """Generate an image, or edit/transform a source image. - - ``image_url`` is the primary source to edit; ``reference_image_urls`` - are extra style/composition references (clamp to ``max_reference_images``). - Any source image routes to the edit endpoint, otherwise text-to-image. - Return :func:`success_response` / :func:`error_response`. Unknown - ``kwargs`` MUST be ignored (forward compat). Known optional kwarg: - ``upscale`` (bool) — a post-generation high-res pass; providers that - honor it report ``upscaled: True`` in ``extra``. - """ - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- + """Generate an image, or edit ``image_url`` (``reference_image_urls`` are extra + style/composition refs, clamped to ``max_reference_images``); any source image + routes to the edit endpoint. Return :func:`success_response` / :func:`error_response`. + Unknown ``kwargs`` MUST be ignored (forward compat); ``upscale`` (bool) is a + post-generation high-res pass, reported as ``upscaled: True`` in ``extra``.""" def resolve_aspect_ratio(value: Optional[str]) -> str: """Clamp to :data:`VALID_ASPECT_RATIOS`; invalid values coerce to landscape so the tool surface forgives agent mistakes instead of rejecting them.""" - if not isinstance(value, str): - return DEFAULT_ASPECT_RATIO - v = value.strip().lower() - if v in VALID_ASPECT_RATIOS: - return v - return DEFAULT_ASPECT_RATIO + v = value.strip().lower() if isinstance(value, str) else "" + return v if v in VALID_ASPECT_RATIOS else DEFAULT_ASPECT_RATIO def normalize_reference_images(value: Any) -> Optional[List[str]]: """Coerce a str or list into a clean list of non-blank strings; ``None`` when nothing usable remains so providers treat "no refs" as one sentinel.""" - if value is None: - return None if isinstance(value, str): value = [value] if not isinstance(value, (list, tuple)): return None - out: List[str] = [] - for item in value: - if isinstance(item, str) and item.strip(): - out.append(item.strip()) - return out or None + return [item.strip() for item in value if isinstance(item, str) and item.strip()] or None -def _images_cache_dir() -> Path: - """Return ``$HERMES_HOME/cache/images/``, creating parents as needed.""" - return provider_media.cache_dir("images") - - -def save_b64_image( - b64_data: str, - *, - prefix: str = "image", - extension: str = "png", -) -> Path: +def save_b64_image(b64_data: str, *, prefix: str = "image", extension: str = "png") -> Path: """Decode base64 image data into ``$HERMES_HOME/cache/images/``; return the path.""" return provider_media.save_b64("images", b64_data, prefix=prefix, extension=extension) _URL_IMAGE_CONTENT_TYPES = { - "image/png": "png", - "image/jpeg": "jpg", - "image/jpg": "jpg", - "image/webp": "webp", - "image/gif": "gif", + "image/png": "png", "image/jpeg": "jpg", "image/jpg": "jpg", "image/webp": "webp", "image/gif": "gif", } def save_url_image( - url: str, - *, - prefix: str = "image", - timeout: float = 60.0, - max_bytes: int = 25 * 1024 * 1024, + url: str, *, prefix: str = "image", timeout: float = 60.0, max_bytes: int = 25 * 1024 * 1024, ) -> Path: - """Download an (often ephemeral) image URL into ``$HERMES_HOME/cache/images/``. - - Raises on network / HTTP / oversize / empty errors so callers can fall back - to returning the bare URL with a clear message. See :mod:`agent.provider_media`. - """ + """Download an (often ephemeral) image URL into ``$HERMES_HOME/cache/images/``. Raises on + network / HTTP / oversize / empty errors so callers can fall back to the bare URL.""" return provider_media.save_url( "images", url, prefix=prefix, timeout=timeout, max_bytes=max_bytes, chunk_size=64 * 1024, content_types=_URL_IMAGE_CONTENT_TYPES, @@ -174,17 +107,11 @@ def success_response( ) -> Dict[str, Any]: """Uniform success dict; ``extra`` keys are added without overriding standard ones.""" payload: Dict[str, Any] = { - "success": True, - "image": image, - "model": model, - "prompt": prompt, - "aspect_ratio": aspect_ratio, - "modality": modality, - "provider": provider, + "success": True, "image": image, "model": model, "prompt": prompt, + "aspect_ratio": aspect_ratio, "modality": modality, "provider": provider, } - if extra: - for k, v in extra.items(): - payload.setdefault(k, v) + for k, v in (extra or {}).items(): + payload.setdefault(k, v) return payload @@ -199,12 +126,6 @@ def error_response( ) -> Dict[str, Any]: """Build a uniform error response dict.""" return { - "success": False, - "image": None, - "error": error, - "error_type": error_type, - "model": model, - "prompt": prompt, - "aspect_ratio": aspect_ratio, - "provider": provider, + "success": False, "image": None, "error": error, "error_type": error_type, + "model": model, "prompt": prompt, "aspect_ratio": aspect_ratio, "provider": provider, } diff --git a/agent/image_gen_registry.py b/agent/image_gen_registry.py index 087144f8be..7c6d2b3365 100644 --- a/agent/image_gen_registry.py +++ b/agent/image_gen_registry.py @@ -1,21 +1,10 @@ -""" -Image Generation Provider Registry -================================== +"""Image generation provider registry. -Central map of registered providers. Populated by plugins at import-time via -``PluginContext.register_image_gen_provider()``; consumed by the -``image_generate`` tool to dispatch each call to the active backend. - -Active selection ----------------- -The active provider is chosen by ``image_gen.provider`` in ``config.yaml``. -If unset, :func:`get_active_provider` applies fallback logic: - -1. If exactly one *available* provider is registered, use it. -2. Otherwise if a provider named ``fal`` is registered and available, use it - (legacy default — matches pre-plugin behavior). -3. Otherwise return ``None`` (the tool surfaces a helpful error pointing - the user at ``hermes tools``). +Populated by plugins at import-time via ``PluginContext.register_image_gen_provider()``; +the ``image_generate`` tool dispatches to :func:`get_active_provider`. Selection is +``image_gen.provider`` in config.yaml; when unset: the single *available* provider, +else ``fal`` if registered and available (legacy default), else ``None`` (the tool +points the user at ``hermes tools``). """ from __future__ import annotations @@ -36,35 +25,23 @@ _registry.export(globals()) def get_active_provider() -> Optional[ImageGenProvider]: - """Resolve the currently-active provider. - - **Availability semantics** (mirrors :mod:`agent.web_search_registry`): - an explicitly configured provider is returned even if ``is_available()`` - is False, so the dispatcher surfaces a precise "X_API_KEY is not set" - error instead of silently switching backends. Only the unconfigured - fallback path is filtered by availability. - """ + """Resolve the currently-active provider. Availability semantics (mirrors + :mod:`agent.web_search_registry`): an explicitly configured provider is returned + even if ``is_available()`` is False, so the dispatcher surfaces a precise + "X_API_KEY is not set" error instead of silently switching backends; only the + unconfigured fallback path is filtered by availability.""" configured = configured_provider_name("image_gen", logger) snapshot = _registry.merged() + if configured: + if snapshot.get(configured) is not None: + return snapshot[configured] + logger.debug("image_gen.provider='%s' configured but not registered; falling back", configured) def _available(p: ImageGenProvider) -> bool: return is_available_safe(p, logger, "image_gen provider %s.is_available() raised %s") - if configured: - provider = snapshot.get(configured) - if provider is not None: - return provider - logger.debug( - "image_gen.provider='%s' configured but not registered; falling back", - configured, - ) - available = [p for p in snapshot.values() if _available(p)] if len(available) == 1: return available[0] - fal = snapshot.get("fal") - if fal is not None and _available(fal): - return fal - - return None + return fal if fal is not None and _available(fal) else None diff --git a/agent/image_routing.py b/agent/image_routing.py index 02c21b9ef6..03de0d79d3 100644 --- a/agent/image_routing.py +++ b/agent/image_routing.py @@ -1,27 +1,12 @@ """Routing helpers for inbound user-attached images. -Two modes: - - native — attach images as OpenAI-style ``image_url`` content parts on the - user turn; provider adapters translate these to vendor formats. - text — run ``vision_analyze`` on each image up-front and prepend the - (lossy) description to the user's text. Still the right choice - for non-vision models. - -:func:`decide_image_input_mode` decides once per message turn from -``agent.image_input_mode`` (``auto`` | ``native`` | ``text``, default ``auto``) -plus the active model's capability metadata. In ``auto`` mode: - - - An explicitly configured ``auxiliary.vision`` backend is the DE-FACTO route - (``text``): a user who named a dedicated vision model wants it used even - when the main model has native vision. ``image_input_mode: native`` is the - absolute override. - - Otherwise, ``supports_vision=True`` (config override or catalog) → native. - - Otherwise text via the default vision_analyze flow. - -``vision_analyze`` stays surfaced as a tool in every session so skills that -chain it keep working; routing only affects how *user-attached images on the -current turn* are presented to the main model. +``native`` attaches images as OpenAI-style ``image_url`` parts; ``text`` runs +``vision_analyze`` up-front and prepends the lossy description (right for +non-vision models). :func:`decide_image_input_mode` picks once per turn from +``agent.image_input_mode`` (``auto`` | ``native`` | ``text``): in ``auto`` an +explicit ``auxiliary.vision`` backend forces ``text`` even for vision-capable +main models (``native`` is the absolute override); else ``supports_vision`` +(config override or catalog) decides. ``vision_analyze`` stays a tool regardless. """ from __future__ import annotations @@ -31,6 +16,8 @@ import logging import mimetypes import os import re +from contextlib import suppress +from io import BytesIO from pathlib import Path from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple @@ -40,99 +27,68 @@ logger = logging.getLogger(__name__) _VALID_MODES = frozenset({"auto", "native", "text"}) -# Extensions extract_image_refs() auto-attaches. Kept tight: documents/archives -# are excluded because the gateway routes them via send_document, and we never -# want a PDF attached as a vision part. -_IMAGE_EXTS = ( - ".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif", ".heic", -) +# Extensions extract_image_refs() auto-attaches. Documents/archives are excluded: +# the gateway routes them via send_document and a PDF must never become a vision part. +_IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif", ".heic") _IMAGE_EXT_PATTERN = "|".join(e.lstrip(".") for e in _IMAGE_EXTS) - -# Absolute / home-relative local image path — same shape as gateway's -# extract_local_files(): anchored to ``~/`` or ``/``, the lookbehind skips -# matches inside URLs, case-insensitive extension. +# Local path: same shape as gateway extract_local_files() — anchored to ``~/`` or +# ``/``, lookbehind skips matches inside URLs. URL: strict ``http(s)://`` so +# ``file://`` and other schemes are not grabbed; optional query string. _LOCAL_IMAGE_PATH_RE = re.compile( - r"(?\"']+?\.(?:" + _IMAGE_EXT_PATTERN + r")(?:\?[^\s<>\"']*)?", - re.IGNORECASE, + r"https?://[^\s<>\"']+?\.(?:" + _IMAGE_EXT_PATTERN + r")(?:\?[^\s<>\"']*)?", re.IGNORECASE, ) +_CODE_SPAN_RES = (re.compile(r"```[^\n]*\n.*?```", re.DOTALL), re.compile(r"`[^`\n]+`")) + + +def _matches_outside_code(pattern: re.Pattern, text: str) -> Iterable[str]: + """Yield ``pattern`` matches whose start is not inside a fenced block or inline backticks.""" + spans = [(m.start(), m.end()) for p in _CODE_SPAN_RES for m in p.finditer(text)] + return (m.group(0) for m in pattern.finditer(text) if not any(s <= m.start() < e for s, e in spans)) + + +def _existing_file(candidate: str) -> Optional[str]: + """Expanded path when it is a regular file; None otherwise (incl. OSError on pathological input).""" + expanded = os.path.expanduser(candidate) + try: + return expanded if os.path.isfile(expanded) else None + except OSError: + return None def extract_image_refs(text: str) -> Tuple[List[str], List[str]]: - """Scan free-form text for image references the model should see. - - Returns ``(local_paths, urls)``, each order-preserving and deduplicated. - Local paths (``/`` or ``~/``) must exist on disk as files; URLs are not - validated (the provider fetches them). Matches inside fenced code blocks - and inline backticks are skipped so pasted example snippets aren't treated - as live attachments (mirrors ``BaseAdapter.extract_local_files``). - """ + """Scan free-form text for image references → ``(local_paths, urls)``, each + ordered and deduplicated. Local paths must exist as files; URLs are not + validated (the provider fetches them). Code spans are skipped so pasted + snippets aren't live attachments (mirrors ``BaseAdapter.extract_local_files``).""" if not isinstance(text, str) or not text: return [], [] - - code_spans: list[tuple[int, int]] = [ - (m.start(), m.end()) - for pattern, flags in ((r"```[^\n]*\n.*?```", re.DOTALL), (r"`[^`\n]+`", 0)) - for m in re.finditer(pattern, text, flags) - ] - - def _in_code(pos: int) -> bool: - return any(s <= pos < e for s, e in code_spans) - - local_paths: list[str] = [] - for match in _LOCAL_IMAGE_PATH_RE.finditer(text): - if _in_code(match.start()): - continue - expanded = os.path.expanduser(match.group(0)) - try: - if not os.path.isfile(expanded): - continue - except OSError: - # ENAMETOOLONG / EINVAL on pathological inputs — skip rather than crash. - continue - if expanded not in local_paths: - local_paths.append(expanded) - - urls: list[str] = [] - for match in _IMAGE_URL_RE.finditer(text): - if _in_code(match.start()): - continue - # Trailing punctuation is almost certainly prose ("see https://x/a.png."). - url = match.group(0).rstrip(".,;:!?)]>") - if url not in urls: - urls.append(url) - - return local_paths, urls + local_paths = dict.fromkeys( + p for p in map(_existing_file, _matches_outside_code(_LOCAL_IMAGE_PATH_RE, text)) if p + ) + # Trailing punctuation is almost certainly prose ("see https://x/a.png."). + urls = dict.fromkeys(u.rstrip(".,;:!?)]>") for u in _matches_outside_code(_IMAGE_URL_RE, text)) + return list(local_paths), list(urls) -# Strict YAML/JSON boolean coercion for capability overrides. ``bool("false")`` -# is True, so a quoted ``supports_vision: "false"`` would silently enable native -# routing on a model that can't handle it. Accept only YAML boolean tokens, -# real bools and 0/1; anything else is None so the caller falls through to -# models.dev rather than honouring garbage. -_TRUE_TOKENS = frozenset({"true", "yes", "on", "1"}) -_FALSE_TOKENS = frozenset({"false", "no", "off", "0"}) +_BOOL_TOKENS = { + **dict.fromkeys(("true", "yes", "on", "1"), True), + **dict.fromkeys(("false", "no", "off", "0"), False), +} def _coerce_capability_bool(raw: Any) -> Optional[bool]: - """Return True/False for recognised boolean values, None otherwise.""" + """Strict boolean coercion for capability overrides: real bools, 0/1 and YAML + boolean tokens only; anything else is None so the caller falls through to + models.dev. ``bool("false")`` is True, so a quoted ``supports_vision: "false"`` + would otherwise silently enable native routing on a model that can't handle it.""" if isinstance(raw, bool): return raw if isinstance(raw, int): return bool(raw) if raw in (0, 1) else None - if isinstance(raw, str): - s = raw.strip().lower() - if s in _TRUE_TOKENS: - return True - if s in _FALSE_TOKENS: - return False - return None + return _BOOL_TOKENS.get(raw.strip().lower()) if isinstance(raw, str) else None def _dict_or_empty(raw: Any) -> Dict[str, Any]: @@ -160,20 +116,17 @@ def _model_supports_vision_override(models_cfg: Any, model: str) -> Optional[boo def _custom_provider_entries(cfg: Dict[str, Any], names: Iterable[str]) -> Iterable[Dict[str, Any]]: - """Yield legacy ``custom_providers`` list entries whose ``name`` matches ``names``. + """Yield legacy ``custom_providers`` entries matching ``names`` (case-insensitive); + ``names`` is the outer loop so list order cannot let a persisted default shadow the live route.""" + entries = _custom_provider_list(cfg) + for wanted in (n.strip().lower() for n in names): + yield from (e for e in entries if _clean_str(e.get("name")).lower() == wanted) - Iterates ``names`` in the given priority order (outer loop) so list order - cannot let a persisted default shadow the live route. - """ - custom_providers = cfg.get("custom_providers") - if not isinstance(custom_providers, list): - return - entries = [e for e in custom_providers if isinstance(e, dict)] - for name in names: - wanted = name.strip().lower() - for entry in entries: - if _clean_str(entry.get("name")).lower() == wanted: - yield entry + +def _custom_provider_list(cfg: Dict[str, Any]) -> List[Dict[str, Any]]: + """Dict entries of the legacy ``custom_providers`` list (empty when absent/malformed).""" + raw = cfg.get("custom_providers") + return [e for e in raw if isinstance(e, dict)] if isinstance(raw, list) else [] def _supports_vision_override( @@ -193,32 +146,22 @@ def _supports_vision_override( """ if not isinstance(cfg, dict): return None - model_cfg = _dict_or_empty(cfg.get("model")) top = _coerce_capability_bool(model_cfg.get("supports_vision")) if top is not None: return top - provider_candidates: List[str] = [] - for candidate in (requested_provider, provider, _clean_str(model_cfg.get("provider"))): - if candidate: - provider_candidates.append(candidate) - if candidate.startswith("custom:") and candidate[len("custom:"):]: - provider_candidates.append(candidate[len("custom:"):]) - provider_candidates = list(dict.fromkeys(provider_candidates)) + candidates: List[str] = [] + for candidate in filter(None, (requested_provider, provider, _clean_str(model_cfg.get("provider")))): + candidates.append(candidate) + if candidate.startswith("custom:") and candidate[len("custom:"):]: + candidates.append(candidate[len("custom:"):]) + candidates = list(dict.fromkeys(candidates)) providers_cfg = _dict_or_empty(cfg.get("providers")) - for p in provider_candidates: - coerced = _model_supports_vision_override(_dict_or_empty(providers_cfg.get(p)).get("models"), model) - if coerced is not None: - return coerced - - for entry in _custom_provider_entries(cfg, provider_candidates): - coerced = _model_supports_vision_override(entry.get("models"), model) - if coerced is not None: - return coerced - - return None + model_maps = [_dict_or_empty(providers_cfg.get(p)).get("models") for p in candidates] + model_maps += [entry.get("models") for entry in _custom_provider_entries(cfg, candidates)] + return next((v for v in (_model_supports_vision_override(m, model) for m in model_maps) if v is not None), None) def _resolve_inference_value( @@ -228,143 +171,131 @@ def _resolve_inference_value( *, runtime_ok: Callable[[str], bool], ) -> str: - """Shared resolution for ``base_url`` / ``api_key`` of the active inference provider. - - Order: context-local runtime value (when ``runtime_ok`` accepts it) → - ``model.`` → ``providers..`` → ``custom_providers[].``, - where ```` covers the provider and ``model.provider`` in both bare - and ``custom:``-prefixed forms. - """ + """``base_url`` / ``api_key`` of the active inference provider. Order: runtime + value (when ``runtime_ok`` accepts it) → ``model.`` → ``providers..`` + → ``custom_providers[].``, ```` covering the provider and + ``model.provider`` in both bare and ``custom:``-prefixed forms.""" runtime = _runtime_main(key) if runtime and runtime_ok(runtime): return runtime - if not isinstance(cfg, dict): return "" - model_cfg = _dict_or_empty(cfg.get("model")) value = _clean_str(model_cfg.get(key)) if value: return value - candidate_names: set[str] = set() + names: set[str] = set() for p in filter(None, (provider, _clean_str(model_cfg.get("provider")))): - candidate_names.add(p) - if p.lower().startswith("custom:"): - candidate_names.add(p.split(":", 1)[1]) - else: - candidate_names.add(f"custom:{p}") - - providers_cfg = cfg.get("providers") - if isinstance(providers_cfg, dict): - for name in candidate_names: - entry = providers_cfg.get(name) - if isinstance(entry, dict): - value = _clean_str(entry.get(key)) - if value: - return value - - custom_providers = cfg.get("custom_providers") - if isinstance(custom_providers, list): - lowered = {n.lower() for n in candidate_names} - for entry_raw in custom_providers: - if not isinstance(entry_raw, dict): - continue - entry_name = _clean_str(entry_raw.get("name")) - if entry_name not in candidate_names and entry_name.lower() not in lowered: - continue - value = _clean_str(entry_raw.get(key)) - if value: - return value - - return "" + names.add(p) + names.add(p.split(":", 1)[1] if p.lower().startswith("custom:") else f"custom:{p}") + lowered = {n.lower() for n in names} + providers_cfg = _dict_or_empty(cfg.get("providers")) + entries = [e for e in map(providers_cfg.get, names) if isinstance(e, dict)] + entries += [ + e for e in _custom_provider_list(cfg) + if _clean_str(e.get("name")) in names or _clean_str(e.get("name")).lower() in lowered + ] + return next((v for v in (_clean_str(e.get(key)) for e in entries) if v), "") -def _resolve_inference_base_url( - cfg: Optional[Dict[str, Any]], - provider: str, -) -> str: - """Best-effort base URL for the active inference provider. - - The runtime base_url is only trusted when it belongs to the requested - provider (or no provider was requested). - """ - requested_provider = _clean_str(provider).lower() - - def _runtime_ok(_: str) -> bool: - return not requested_provider or requested_provider == _runtime_main("provider").lower() - - return _resolve_inference_value(cfg, provider, "base_url", runtime_ok=_runtime_ok) +def _resolve_inference_base_url(cfg: Optional[Dict[str, Any]], provider: str) -> str: + """Best-effort base URL for the active inference provider; the runtime value is + only trusted when it belongs to the requested provider (or none was requested).""" + requested = _clean_str(provider).lower() + return _resolve_inference_value( + cfg, provider, "base_url", + runtime_ok=lambda _: not requested or requested == _runtime_main("provider").lower(), + ) -def _resolve_inference_api_key( - cfg: Optional[Dict[str, Any]], - provider: str, -) -> str: - """Best-effort API key for the active inference provider. - - Mirrors :func:`_resolve_inference_base_url` so the key matches the base URL - actually probed; otherwise the local server-type probe hits a keyed remote - endpoint without Authorization and sprays 401s on every image turn. - """ +def _resolve_inference_api_key(cfg: Optional[Dict[str, Any]], provider: str) -> str: + """Best-effort API key, resolved like :func:`_resolve_inference_base_url` so it + matches the base URL actually probed; otherwise the local server-type probe hits + a keyed remote endpoint without Authorization and sprays 401s on every image turn.""" return _resolve_inference_value(cfg, provider, "api_key", runtime_ok=lambda _: True) -def _should_probe_ollama_vision( - provider: str, base_url: str, api_key: str = "" -) -> bool: - """True when the active provider likely fronts a local Ollama server. - - Server-fingerprint probing is only valid for LOCAL endpoints: remote - OpenAI-compatible APIs (sglang, vLLM) expose Ollama-compat routes that can - misidentify, and probing them without an api_key returns 401 on every leg. - """ - if (provider or "").strip().lower() == "ollama": +def _should_probe_ollama_vision(provider: str, base_url: str, api_key: str = "") -> bool: + """True when the active provider likely fronts a local Ollama server. Fingerprint + probing is only valid for LOCAL endpoints: remote OpenAI-compatible APIs (sglang, + vLLM) expose Ollama-compat routes that can misidentify, and probing them without + an api_key returns 401 on every leg.""" + if _clean_str(provider).lower() == "ollama": return True if not base_url: return False try: from agent.model_metadata import detect_local_server_type, is_local_endpoint - if not is_local_endpoint(base_url): - return False # Forward the key: an unauthorized probe can never produce a positive verdict. - return detect_local_server_type(base_url, api_key=api_key) == "ollama" + return bool(is_local_endpoint(base_url)) and detect_local_server_type(base_url, api_key=api_key) == "ollama" except Exception: return False def _coerce_mode(raw: Any) -> str: """Normalize a config value into one of the valid modes (default ``auto``).""" - if isinstance(raw, str) and raw.strip().lower() in _VALID_MODES: - return raw.strip().lower() - return "auto" + mode = raw.strip().lower() if isinstance(raw, str) else "" + return mode if mode in _VALID_MODES else "auto" def _explicit_aux_vision_override(cfg: Optional[Dict[str, Any]]) -> bool: - """True when the user configured a specific ``auxiliary.vision`` backend. - - An explicit backend is the DE-FACTO image route in ``auto`` mode even when - the main model could take images natively. ``auto``/empty provider with no - model and no base_url is not explicit. - """ - if not isinstance(cfg, dict): - return False - aux = cfg.get("auxiliary") or {} - if not isinstance(aux, dict): - return False - vision = aux.get("vision") or {} - if not isinstance(vision, dict): - return False - - provider = _clean_str(vision.get("provider")).lower() - return not ( - provider in {"", "auto"} + """True when the user configured a specific ``auxiliary.vision`` backend — the + de-facto image route in ``auto`` mode even when the main model has native vision. + ``auto``/empty provider with no model and no base_url is not explicit.""" + vision = _dict_or_empty(_dict_or_empty(_dict_or_empty(cfg).get("auxiliary")).get("vision")) + return bool(vision) and not ( + _clean_str(vision.get("provider")).lower() in {"", "auto"} and not _clean_str(vision.get("model")) and not _clean_str(vision.get("base_url")) ) +def _probe_managed_runtime(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> Optional[bool]: + """Managed local runtime verdict: the server receiving the image is the authority + on whether it can see (its /props reports modalities). Cloud catalogs have never + heard of a local GGUF, so without this every local model reads as text-only and + screenshots detour to a cloud auxiliary.""" + from hermes_cli.local_runtime.capabilities import is_managed_provider, managed_model_supports_vision + + managed = is_managed_provider(provider, _resolve_inference_base_url(cfg, provider) or "") + return managed_model_supports_vision(model) if managed else None + + +def _probe_models_dev(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> Optional[bool]: + """models.dev catalog verdict. ``allow_network=True`` on purpose: this runs only + when an image needs routing, and the text-only-main guard depends on catalog + data — a cold cache returning "unknown" would reintroduce attempting the call. + The fetch is cached (4h TTL) and backoff-limited.""" + from agent.models_dev import get_model_capabilities + + caps = get_model_capabilities(provider, model, allow_network=True) + return None if caps is None else bool(caps.supports_vision) + + +def _probe_ollama(provider: str, model: str, cfg: Optional[Dict[str, Any]]) -> Optional[bool]: + """Ollama ``/api/show`` verdict for local endpoints (see :func:`_should_probe_ollama_vision`).""" + base_url = _resolve_inference_base_url(cfg, provider) + if not base_url and _clean_str(provider).lower() == "ollama": + base_url = "http://localhost:11434/v1" + api_key = _resolve_inference_api_key(cfg, provider) + if not _should_probe_ollama_vision(provider, base_url, api_key=api_key): + return None + from agent.model_metadata import query_ollama_supports_vision + + return query_ollama_supports_vision(model, base_url, api_key=api_key) + + +# Capability probes after the config override, in priority order; each returns +# True/False or None (unknown → next probe). Exceptions are logged and treated as None. +_VISION_PROBES: Tuple[Tuple[str, Callable[..., Optional[bool]]], ...] = ( + ("managed-runtime caps lookup", _probe_managed_runtime), + ("caps lookup", _probe_models_dev), + ("ollama vision probe", _probe_ollama), +) + + def _lookup_supports_vision( provider: str, model: str, @@ -374,84 +305,34 @@ def _lookup_supports_vision( ) -> Optional[bool]: """Return True/False if vision capability can be resolved, None if unknown. - Order: config ``supports_vision`` override → managed local runtime → - models.dev catalog → Ollama probe for local endpoints. + Order: config ``supports_vision`` override → :data:`_VISION_PROBES` + (managed local runtime → models.dev catalog → Ollama probe). """ # Named custom providers are canonicalized to ``provider="custom"``; the # original name lives in the context-local main runtime. Borrow it only on an # exact provider+model match so background/auxiliary lookups never take # another turn's identity. - if not requested_provider: - if ( - _runtime_main("provider").lower() == _clean_str(provider).lower() - and _runtime_main("model") == _clean_str(model) - ): - requested_provider = _runtime_main("requested_provider") + if ( + not requested_provider + and _runtime_main("provider").lower() == _clean_str(provider).lower() + and _runtime_main("model") == _clean_str(model) + ): + requested_provider = _runtime_main("requested_provider") - override = _supports_vision_override( - cfg, - provider, - model, - requested_provider=requested_provider, - ) + override = _supports_vision_override(cfg, provider, model, requested_provider=requested_provider) if override is not None: return override if not provider or not model: return None - # Managed local runtime: the server receiving the image is the authority on - # whether it can see (its /props reports modalities). Cloud catalogs have - # never heard of a local GGUF, so without this every local model reads as - # text-only and screenshots detour to a cloud auxiliary. - try: - from hermes_cli.local_runtime.capabilities import ( - is_managed_provider, - managed_model_supports_vision, - ) - - if is_managed_provider(provider, _resolve_inference_base_url(cfg, provider) or ""): - managed = managed_model_supports_vision(model) - if managed is not None: - return managed - except Exception as exc: # pragma: no cover - defensive - logger.debug("image_routing: managed-runtime caps lookup failed for %s:%s — %s", - provider, model, exc) - - caps = None - try: - from agent.models_dev import get_model_capabilities - # allow_network=True on purpose: this runs only when an image needs - # routing, and the text-only-main guard depends on catalog data — a cold - # cache returning "unknown" would reintroduce attempting the call. The - # fetch is cached (4h TTL) and backoff-limited. - caps = get_model_capabilities(provider, model, allow_network=True) - except Exception as exc: # pragma: no cover - defensive - logger.debug("image_routing: caps lookup failed for %s:%s — %s", provider, model, exc) - if caps is not None: - return bool(caps.supports_vision) - - base_url = _resolve_inference_base_url(cfg, provider) - if not base_url and (provider or "").strip().lower() == "ollama": - base_url = "http://localhost:11434/v1" - - resolved_api_key = _resolve_inference_api_key(cfg, provider) - - if _should_probe_ollama_vision(provider, base_url, api_key=resolved_api_key): + for label, probe in _VISION_PROBES: try: - from agent.model_metadata import query_ollama_supports_vision - - ollama_vision = query_ollama_supports_vision( - model, base_url, api_key=resolved_api_key - ) - if ollama_vision is not None: - return ollama_vision + verdict = probe(provider, model, cfg) except Exception as exc: # pragma: no cover - defensive - logger.debug( - "image_routing: ollama vision probe failed for %s:%s — %s", - provider, - model, - exc, - ) + logger.debug("image_routing: %s failed for %s:%s — %s", label, provider, model, exc) + continue + if verdict is not None: + return verdict return None @@ -462,61 +343,40 @@ def decide_image_input_mode( *, requested_provider: str = "", ) -> str: - """Return ``"native"`` or ``"text"`` for the given turn. - - Args: - provider: active inference provider ID (e.g. ``"anthropic"``). - model: active model slug as sent to the provider. - cfg: loaded config.yaml dict, or None (behaves as auto). - requested_provider: provider identity before runtime canonicalization. - """ - mode_cfg = "auto" - if isinstance(cfg, dict): - agent_cfg = cfg.get("agent") or {} - if isinstance(agent_cfg, dict): - mode_cfg = _coerce_mode(agent_cfg.get("image_input_mode")) - + """Return ``"native"`` or ``"text"`` for the given turn (``cfg`` None behaves as + auto; ``requested_provider`` is the identity before runtime canonicalization).""" + mode_cfg = _coerce_mode(_dict_or_empty(_dict_or_empty(cfg).get("agent")).get("image_input_mode")) if mode_cfg != "auto": return mode_cfg - - # auto: an explicit auxiliary.vision backend wins (see module docstring); - # native remains the default for unconfigured installs. - if _explicit_aux_vision_override(cfg): + if _explicit_aux_vision_override(cfg): # auto: an explicit auxiliary.vision backend wins return "text" - if requested_provider: - supports = _lookup_supports_vision( - provider, - model, - cfg, - requested_provider=requested_provider, - ) - else: - # Keep the three-argument call contract for callers/tests that replace - # the capability lookup hook. - supports = _lookup_supports_vision(provider, model, cfg) - return "native" if supports is True else "text" + # Keep the three-argument call contract for callers/tests that replace the lookup hook. + extra = {"requested_provider": requested_provider} if requested_provider else {} + return "native" if _lookup_supports_vision(provider, model, cfg, **extra) is True else "text" -# Image size handling is REACTIVE: attach at full size regardless of provider -# and let ``run_agent._try_shrink_image_parts_in_messages`` shrink + retry on -# rejection (e.g. Anthropic's 5 MB ceiling as HTTP 400). Provider ceilings are -# partial and evolving (OpenAI 49 MB+, Anthropic 5 MB, Gemini 100 MB); a -# proactive table would go stale and silently degrade quality for providers -# that would have accepted the full image — worse than one extra API call. +# Image size handling is REACTIVE: attach at full size and let +# ``run_agent._try_shrink_image_parts_in_messages`` shrink + retry on rejection +# (e.g. Anthropic's 5 MB ceiling as HTTP 400). Provider ceilings are partial and +# evolving; a proactive table would go stale and silently degrade quality. - -# Magic-byte signatures, checked in order. Filename-based detection is -# unreliable when platforms lie about content-type (Discord serves PNG as -# ``image/webp`` for proxied stickers); Anthropic rejects a media_type that -# does not match the bytes with HTTP 400, so we sniff. -_HEIC_BRANDS = frozenset({ - b"heic", b"heix", b"hevc", b"hevx", b"mif1", b"msf1", b"heim", b"heis", -}) -_MAGIC_PREFIXES: Tuple[Tuple[bytes, str], ...] = ( - (b"\x89PNG\r\n\x1a\n", "image/png"), - (b"\xff\xd8\xff", "image/jpeg"), - (b"GIF87a", "image/gif"), - (b"GIF89a", "image/gif"), +# Magic-byte signatures as ((offset, bytes), ...) conjunctions, checked in order. +# Platforms lie about content-type (Discord serves PNG as ``image/webp`` for +# proxied stickers) and Anthropic rejects a mismatched media_type with HTTP 400. +# ISO-BMFF family (HEIC/HEIF/AVIF): 'ftyp' at 4..8, major brand at 8..12. +_FTYP_BRANDS = { + **dict.fromkeys((b"avif", b"avis"), "image/avif"), + **dict.fromkeys((b"heic", b"heix", b"hevc", b"hevx", b"mif1", b"msf1", b"heim", b"heis"), "image/heic"), +} +_MAGIC: Tuple[Tuple[Tuple[Tuple[int, bytes], ...], str], ...] = ( + (((0, b"\x89PNG\r\n\x1a\n"),), "image/png"), + (((0, b"\xff\xd8\xff"),), "image/jpeg"), + (((0, b"GIF87a"),), "image/gif"), (((0, b"GIF89a"),), "image/gif"), + (((0, b"RIFF"), (8, b"WEBP")), "image/webp"), + (((0, b"BM"),), "image/bmp"), + *((((4, b"ftyp"), (8, brand)), mime) for brand, mime in _FTYP_BRANDS.items()), + (((0, b"II*\x00"),), "image/tiff"), (((0, b"MM\x00*"),), "image/tiff"), + (((0, b"\x00\x00\x01\x00"),), "image/x-icon"), ) @@ -524,48 +384,25 @@ def _sniff_mime_from_bytes(raw: bytes) -> Optional[str]: """Detect image MIME from magic bytes; None if unrecognised.""" if not raw: return None - for prefix, mime in _MAGIC_PREFIXES: - if raw.startswith(prefix): + for conditions, mime in _MAGIC: + if all(raw[off:off + len(sig)] == sig for off, sig in conditions): return mime - if len(raw) >= 12 and raw[:4] == b"RIFF" and raw[8:12] == b"WEBP": - return "image/webp" - if raw.startswith(b"BM"): - return "image/bmp" - # ISO-BMFF family (HEIC/HEIF/AVIF): 'ftyp' at 4..8, major brand at 8..12. - if len(raw) >= 12 and raw[4:8] == b"ftyp": - brand = raw[8:12] - if brand in {b"avif", b"avis"}: - return "image/avif" - if brand in _HEIC_BRANDS: - return "image/heic" - if raw[:4] in {b"II*\x00", b"MM\x00*"}: - return "image/tiff" - if raw[:4] == b"\x00\x00\x01\x00": - return "image/x-icon" # SVG is text: look for an Optional[bytes]: - """Decode image bytes with Pillow and re-encode as PNG; None when impossible. - - HEIC/HEIF and AVIF need optional Pillow plugins, registered on demand; a - missing plugin just looks like "Pillow can't decode this" so the caller - skips the image and the rest of the turn proceeds. - """ + """Decode with Pillow and re-encode as PNG; None when impossible. HEIC/HEIF and + AVIF need optional Pillow plugins, registered on demand; a missing plugin just + looks like "can't decode" so the caller skips the image and the turn proceeds.""" try: from PIL import Image except ImportError: @@ -575,66 +412,61 @@ def _transcode_to_png(raw: bytes) -> Optional[bytes]: "(and `pillow-heif` / `pillow-avif-plugin` for those formats)." ) return None - try: + with suppress(Exception): import pillow_heif # type: ignore pillow_heif.register_heif_opener() - except Exception: - pass - try: + with suppress(Exception): import pillow_avif # type: ignore # noqa: F401 -- registers AVIF on import - except Exception: - pass try: - from io import BytesIO - with Image.open(BytesIO(raw)) as im: - # Normalise exotic modes to RGBA so PNG can serialise and any - # source transparency survives. + # Normalise exotic modes to RGBA so PNG can serialise and transparency survives. if im.mode not in {"RGB", "RGBA", "L", "LA", "P"}: im = im.convert("RGBA") buf = BytesIO() im.save(buf, format="PNG", optimize=False) return buf.getvalue() except Exception as exc: - logger.info( - "image_routing: Pillow could not transcode image to PNG -- %s", exc - ) + logger.info("image_routing: Pillow could not transcode image to PNG -- %s", exc) return None _SUFFIX_MIMES = { - ".jpg": "image/jpeg", - ".jpeg": "image/jpeg", - ".png": "image/png", - ".gif": "image/gif", - ".webp": "image/webp", - ".bmp": "image/bmp", + ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", + ".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp", } def _guess_mime(path: Path, raw: Optional[bytes] = None) -> str: """Image MIME for *path*: magic bytes (authoritative) → ``mimetypes`` → suffix → jpeg.""" - if raw is not None: - sniffed = _sniff_mime_from_bytes(raw) - if sniffed: - return sniffed - mime, _ = mimetypes.guess_type(str(path)) + sniffed = _sniff_mime_from_bytes(raw) if raw is not None else None + mime = sniffed or mimetypes.guess_type(str(path))[0] if mime and mime.startswith("image/"): return mime # mimetypes on some Linux distros mis-maps .jpg; default to jpeg. return _SUFFIX_MIMES.get(path.suffix.lower(), "image/jpeg") -def _file_to_data_url(path: Path) -> Optional[str]: - """Encode a local image as a base64 data URL at its native size. +def _accepted_mimes() -> frozenset: + """Provider-accepted MIME set for the current main runtime. The managed local + server decodes fewer formats (no WebP — and a WebP part fails SILENTLY: the model + confabulates a description), so its narrower set transcodes those here.""" + try: + from agent.auxiliary_client import _runtime_main_value + from hermes_cli.local_runtime.capabilities import ACCEPTED_IMAGE_MIMES, is_managed_provider - Size is NOT limited here (the agent retry loop shrinks on the provider's - first rejection, so lenient providers pay no silent quality tax). Format - compatibility IS handled: MIMEs outside the accepted set are transcoded to - PNG. Returns None when the file can't be read, is blocked by the read - guard, or can't be transcoded — the caller reports it in ``skipped``. - """ + if is_managed_provider(str(_runtime_main_value("provider") or ""), str(_runtime_main_value("base_url") or "")): + return ACCEPTED_IMAGE_MIMES + except Exception: # noqa: BLE001 — best-effort narrowing only + pass + return _UNIVERSALLY_SUPPORTED_MIMES + + +def _file_to_data_url(path: Path) -> Optional[str]: + """Encode a local image as a base64 data URL at native size (the agent retry + loop shrinks on rejection, so lenient providers pay no silent quality tax); + MIMEs outside the accepted set are transcoded to PNG. None when unreadable, + blocked by the read guard, or untranscodable — the caller reports ``skipped``.""" try: from agent.file_safety import raise_if_read_blocked @@ -643,50 +475,23 @@ def _file_to_data_url(path: Path) -> Optional[str]: logger.warning("image_routing: blocked local image attachment %s -- %s", path, exc) return None except Exception: - # Keep attachment routing best-effort if the guard itself is unavailable. - pass - + pass # Keep attachment routing best-effort if the guard itself is unavailable. try: raw = path.read_bytes() except Exception as exc: logger.warning("image_routing: failed to read %s — %s", path, exc) return None mime = _guess_mime(path, raw=raw) - accepted = _UNIVERSALLY_SUPPORTED_MIMES - # The managed local server decodes fewer formats (no WebP — and a WebP part - # fails SILENTLY: the model confabulates a description). Narrow the accepted - # set so those formats transcode here instead of vanishing server-side. - 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 "")): - accepted = ACCEPTED_IMAGE_MIMES - except Exception: # noqa: BLE001 — best-effort narrowing only - pass - if mime not in accepted: - transcoded = _transcode_to_png(raw) - if transcoded is None: + if mime not in _accepted_mimes(): + if (transcoded := _transcode_to_png(raw)) is None: logger.warning( - "image_routing: %s is %s which is not accepted by the " - "active provider and could not be transcoded to PNG; " - "skipping this attachment.", - path, mime, + "image_routing: %s is %s which is not accepted by the active provider " + "and could not be transcoded to PNG; skipping this attachment.", path, mime, ) return None - logger.info( - "image_routing: transcoded %s (%s) -> image/png for provider compatibility", - path.name, mime, - ) - raw = transcoded - mime = "image/png" - b64 = base64.b64encode(raw).decode("ascii") - return f"data:{mime};base64,{b64}" + logger.info("image_routing: transcoded %s (%s) -> image/png for provider compatibility", path.name, mime) + raw, mime = transcoded, "image/png" + return f"data:{mime};base64,{base64.b64encode(raw).decode('ascii')}" def build_native_content_parts( @@ -696,49 +501,30 @@ def build_native_content_parts( ) -> Tuple[List[Dict[str, Any]], List[str]]: """Build an OpenAI-style ``content`` list for a user turn. - Local paths are embedded as base64 ``data:`` URLs; remote URLs pass through - verbatim. When at least one image attaches, a single text part combines the - caption (or a neutral default) with one hint per image — - ``[Image attached at: ]`` / ``[Image attached: ]`` — giving the - model a string handle for tools that take an image path/URL, mirroring the - text-mode hint from ``Runner._enrich_message_with_vision``. - - Returns ``(content_parts, skipped)``; ``skipped`` holds local paths that - could not be read. URLs are never skipped (not validated here). + Local paths become base64 ``data:`` URLs; remote URLs pass through verbatim. + When any image attaches, one text part combines the caption (or a neutral + default) with a ``[Image attached at: ]`` / ``[Image attached: ]`` + hint per image — a string handle for tools taking an image path/URL, mirroring + ``Runner._enrich_message_with_vision``. Returns ``(content_parts, skipped)``; + ``skipped`` holds unreadable local paths (URLs are never skipped). """ skipped: List[str] = [] - image_parts: List[Dict[str, Any]] = [] - hint_lines: List[str] = [] - + attached: List[Tuple[str, str]] = [] # (url, hint) for raw_path in image_paths: p = Path(raw_path) data_url = _file_to_data_url(p) if p.exists() and p.is_file() else None - if not data_url: + if data_url: + attached.append((data_url, f"[Image attached at: {raw_path}]")) + else: skipped.append(str(raw_path)) - continue - image_parts.append({"type": "image_url", "image_url": {"url": data_url}}) - hint_lines.append(f"[Image attached at: {raw_path}]") - - for url in image_urls or []: - url = (url or "").strip() - if not url: - continue - image_parts.append({"type": "image_url", "image_url": {"url": url}}) - hint_lines.append(f"[Image attached: {url}]") + attached += [(u, f"[Image attached: {u}]") for u in ((u or "").strip() for u in image_urls or []) if u] text = (user_text or "").strip() - - if image_parts: - base_text = text or "What do you see in this image?" - combined_text = f"{base_text}\n\n" + "\n".join(hint_lines) - return [{"type": "text", "text": combined_text}, *image_parts], skipped - - # No images attached — plain text-only behaviour. - return ([{"type": "text", "text": text}] if text else []), skipped + if not attached: + return ([{"type": "text", "text": text}] if text else []), skipped + combined_text = f"{text or 'What do you see in this image?'}\n\n" + "\n".join(h for _, h in attached) + image_parts = [{"type": "image_url", "image_url": {"url": u}} for u, _ in attached] + return [{"type": "text", "text": combined_text}, *image_parts], skipped -__all__ = [ - "decide_image_input_mode", - "build_native_content_parts", - "extract_image_refs", -] +__all__ = ["decide_image_input_mode", "build_native_content_parts", "extract_image_refs"] diff --git a/agent/inline_tool_executors.py b/agent/inline_tool_executors.py index 83108bba69..1a7f603e88 100644 --- a/agent/inline_tool_executors.py +++ b/agent/inline_tool_executors.py @@ -4,14 +4,15 @@ These tools need live ``AIAgent`` state (stores, callbacks, session DB) and ther bypass the tool registry. Each executor is ``fn(agent, args, ctx) -> result``; the table replaces two hand-maintained if/elif chains (``invoke_tool`` and ``execute_tool_calls_sequential``) that had drifted apart. Tool modules are imported -lazily inside the bodies so ``patch("tools.x.y")`` in tests keeps working. +lazily at call time so ``patch("tools.x.y")`` in tests keeps working. """ from __future__ import annotations import json from dataclasses import dataclass -from typing import Any, Callable, Dict, Optional +from importlib import import_module +from typing import Any, Callable, Dict, Optional, Tuple def tool_hook_ids(agent, effective_task_id: str, tool_call_id: Optional[str]) -> Dict[str, str]: @@ -66,27 +67,30 @@ class InlineToolContext: messages: Optional[list] = None -def _todo_list(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.todo_tool import todo_tool as _todo_tool +InlineToolExecutor = Callable[[Any, dict, InlineToolContext], Any] - return _todo_tool( - todos=args.get("todos"), - merge=args.get("merge", False), - store=agent._todo_store, - ) +# ``(kwarg, args_key)`` → ``args.get(key)``; ``(kwarg, args_key, default)`` → ``args.get(key, default)``. +_ArgSpec = Tuple[Any, ...] -def _message_agent(agent, args: dict, ctx: InlineToolContext) -> Any: - # Bot Mode teammate DM is injected, not registered: only a canonical Bot - # Chat session carries the schema, and the tool re-gates on the title. - from tools.bot_mode_dm import message_agent_tool as _message_agent_tool +def _call_tool(module: str, func: str, args: dict, arg_specs: Tuple[_ArgSpec, ...], **fixed: Any) -> Any: + """Import ``module.func`` lazily and call it with args mapped per ``arg_specs`` plus ``fixed``.""" + fn = getattr(import_module(module), func) + return fn(**{spec[0]: args.get(*spec[1:]) for spec in arg_specs}, **fixed) - return _message_agent_tool( - target=args.get("target", ""), - message=args.get("message", ""), - task_id=ctx.effective_task_id, - agent=agent, - ) + +def _tool( + module: str, func: str, *arg_specs: _ArgSpec, **fixed: Callable[[Any, InlineToolContext], Any], +) -> InlineToolExecutor: + """Executor calling ``module.func`` with mapped args plus ``fixed`` kwargs computed from ``(agent, ctx)``.""" + def _exec(agent, args: dict, ctx: InlineToolContext) -> Any: + return _call_tool(module, func, args, arg_specs, **{k: f(agent, ctx) for k, f in fixed.items()}) + return _exec + + +def _callback_tool(module: str, func: str, callback_attr: str, *arg_specs: _ArgSpec) -> InlineToolExecutor: + """Executor for a GUI-callback tool: mapped args plus ``callback=getattr(agent, callback_attr, None)``.""" + return _tool(module, func, *arg_specs, callback=lambda agent, ctx: getattr(agent, callback_attr, None)) def _session_search(agent, args: dict, ctx: InlineToolContext) -> Any: @@ -95,31 +99,24 @@ def _session_search(agent, args: dict, ctx: InlineToolContext) -> Any: from hermes_state import format_session_db_unavailable return json.dumps({"success": False, "error": format_session_db_unavailable()}) - from tools.session_search_tool import session_search as _session_search_tool - - return _session_search_tool( - query=args.get("query", ""), - role_filter=args.get("role_filter"), - limit=args.get("limit", 3), - session_id=args.get("session_id"), - around_message_id=args.get("around_message_id"), - window=args.get("window", 5), - sort=args.get("sort"), - detail=args.get("detail", "adaptive"), - db=session_db, - current_session_id=agent.session_id, + return _call_tool( + "tools.session_search_tool", "session_search", args, + ( + ("query", "query", ""), ("role_filter", "role_filter"), ("limit", "limit", 3), + ("session_id", "session_id"), ("around_message_id", "around_message_id"), + ("window", "window", 5), ("sort", "sort"), ("detail", "detail", "adaptive"), + ), + db=session_db, current_session_id=agent.session_id, ) def _memory(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.memory_tool import memory_tool as _memory_tool - - result = _memory_tool( - action=args.get("action"), - target=args.get("target", "memory"), - content=args.get("content"), - old_text=args.get("old_text"), - operations=args.get("operations"), + result = _call_tool( + "tools.memory_tool", "memory_tool", args, + ( + ("action", "action"), ("target", "target", "memory"), ("content", "content"), + ("old_text", "old_text"), ("operations", "operations"), + ), store=agent._memory_store, ) # Mirror built-in memory writes to external providers; gating lives in @@ -136,129 +133,69 @@ def _memory(agent, args: dict, ctx: InlineToolContext) -> Any: return result -def _clarify(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.clarify_tool import clarify_tool as _clarify_tool - - return _clarify_tool( - question=args.get("question", ""), - choices=args.get("choices"), - multi_select=args.get("multi_select", False), - questions=args.get("questions"), - callback=agent.clarify_callback, - ) - - -def _read_terminal(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.read_terminal_tool import read_terminal_tool as _read_terminal_tool - - return _read_terminal_tool( - start_line=args.get("start_line"), - count=args.get("count"), - callback=getattr(agent, "read_terminal_callback", None), - ) +_read_preview = _callback_tool( + "tools.read_preview_tool", "read_preview_tool", "read_preview_callback", + ("start", "start"), ("count", "count"), +) def _desktop_preview(agent, args: dict, ctx: InlineToolContext) -> Any: # action=read needs the GUI callback (agent-level); open/close go through the # registry handler like any other tool. if (args.get("action") or "").strip() == "read": - from tools.read_preview_tool import read_preview_tool as _read_preview_tool - - return _read_preview_tool( - start=args.get("start"), - count=args.get("count"), - callback=getattr(agent, "read_preview_callback", None), - ) + return _read_preview(agent, args, ctx) from tools.preview_tool import _handle_preview return _handle_preview(args) -def _drive_preview(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.drive_preview_tool import drive_preview_tool as _drive_preview_tool - - return _drive_preview_tool( - action=args.get("action", ""), - ref=args.get("ref"), - selector=args.get("selector"), - text=args.get("text"), - key=args.get("key"), - submit=args.get("submit"), - amount=args.get("amount"), - to=args.get("to"), - limit=args.get("max"), - callback=getattr(agent, "drive_preview_callback", None), - ) - - -def _annotate_preview(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.annotate_preview_tool import annotate_preview_tool as _annotate_preview_tool - - return _annotate_preview_tool( - action=args.get("action", "add"), - ref=args.get("ref"), - selector=args.get("selector"), - label=args.get("label"), - callback=getattr(agent, "drive_preview_callback", None), - ) - - -def _read_window_below(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.read_window_tool import read_window_below_tool as _read_window_below_tool - - return _read_window_below_tool( - callback=getattr(agent, "read_window_below_callback", None), - ) - - -def _gui_tour(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.tour_tool import tour_tool as _tour_tool - - return _tour_tool( - action=args.get("action", ""), - surface=args.get("surface"), - selector=args.get("selector"), - title=args.get("title"), - text=args.get("text"), - side=args.get("side"), - steps=args.get("steps"), - step_index=args.get("step_index"), - callback=getattr(agent, "tour_callback", None), - ) - - -def _setup_mcp(agent, args: dict, ctx: InlineToolContext) -> Any: - from tools.setup_mcp_tool import setup_mcp_tool as _setup_mcp_tool - - return _setup_mcp_tool( - server=args.get("server", ""), - action=args.get("action", "install"), - reason=args.get("reason", ""), - callback=getattr(agent, "setup_mcp_callback", None), - ) - - -def _delegate_task(agent, args: dict, ctx: InlineToolContext) -> Any: - return agent._dispatch_delegate_task(args) - - -InlineToolExecutor = Callable[[Any, dict, InlineToolContext], Any] - # Order is the historical if/elif order of ``execute_tool_calls_sequential``. INLINE_TOOL_EXECUTORS: Dict[str, InlineToolExecutor] = { - "todo_list": _todo_list, - "message_agent": _message_agent, + "todo_list": _tool( + "tools.todo_tool", "todo_tool", ("todos", "todos"), ("merge", "merge", False), + store=lambda agent, ctx: agent._todo_store, + ), + # Bot Mode teammate DM is injected, not registered: only a canonical Bot + # Chat session carries the schema, and the tool re-gates on the title. + "message_agent": _tool( + "tools.bot_mode_dm", "message_agent_tool", ("target", "target", ""), ("message", "message", ""), + task_id=lambda agent, ctx: ctx.effective_task_id, agent=lambda agent, ctx: agent, + ), "session_search": _session_search, "memory": _memory, - "clarify": _clarify, - "read_terminal": _read_terminal, + "clarify": _tool( + "tools.clarify_tool", "clarify_tool", + ("question", "question", ""), ("choices", "choices"), ("multi_select", "multi_select", False), + ("questions", "questions"), + callback=lambda agent, ctx: agent.clarify_callback, + ), + "read_terminal": _callback_tool( + "tools.read_terminal_tool", "read_terminal_tool", "read_terminal_callback", + ("start_line", "start_line"), ("count", "count"), + ), "desktop_preview": _desktop_preview, - "drive_preview": _drive_preview, - "annotate_preview": _annotate_preview, - "read_window_below": _read_window_below, - "gui_tour": _gui_tour, - "setup_mcp": _setup_mcp, - "delegate_task": _delegate_task, + "drive_preview": _callback_tool( + "tools.drive_preview_tool", "drive_preview_tool", "drive_preview_callback", + ("action", "action", ""), ("ref", "ref"), ("selector", "selector"), ("text", "text"), + ("key", "key"), ("submit", "submit"), ("amount", "amount"), ("to", "to"), ("limit", "max"), + ), + "annotate_preview": _callback_tool( + "tools.annotate_preview_tool", "annotate_preview_tool", "drive_preview_callback", + ("action", "action", "add"), ("ref", "ref"), ("selector", "selector"), ("label", "label"), + ), + "read_window_below": _callback_tool( + "tools.read_window_tool", "read_window_below_tool", "read_window_below_callback", + ), + "gui_tour": _callback_tool( + "tools.tour_tool", "tour_tool", "tour_callback", + ("action", "action", ""), ("surface", "surface"), ("selector", "selector"), ("title", "title"), + ("text", "text"), ("side", "side"), ("steps", "steps"), ("step_index", "step_index"), + ), + "setup_mcp": _callback_tool( + "tools.setup_mcp_tool", "setup_mcp_tool", "setup_mcp_callback", + ("server", "server", ""), ("action", "action", "install"), ("reason", "reason", ""), + ), + "delegate_task": lambda agent, args, ctx: agent._dispatch_delegate_task(args), } # ``invoke_tool`` (concurrent path) consults the memory manager right after these three diff --git a/agent/kanban_stop.py b/agent/kanban_stop.py index 4cedc78687..5f4669bdb8 100644 --- a/agent/kanban_stop.py +++ b/agent/kanban_stop.py @@ -1,9 +1,7 @@ -"""Turn-end guard for kanban workers. - -Kanban workers must end with ``kanban_complete`` or ``kanban_block``. Some -models narrate the next step and stop with no tool calls; Hermes treats that -as a clean exit → ``rc=0`` → dispatcher ``protocol_violation``. Policy-only: -return a bounded synthetic nudge so the loop continues instead of exiting. +"""Turn-end guard for kanban workers, which must end with ``kanban_complete`` or +``kanban_block``. Some models narrate the next step and stop with no tool calls; +Hermes treats that as a clean exit → ``rc=0`` → dispatcher ``protocol_violation``. +Policy-only: return a bounded synthetic nudge so the loop continues instead of exiting. """ from __future__ import annotations @@ -19,36 +17,29 @@ _DEFAULT_MAX_ATTEMPTS = 2 def kanban_stop_nudge_enabled() -> bool: """On when ``HERMES_KANBAN_TASK`` is set, unless ``HERMES_KANBAN_STOP_NUDGE`` disables it.""" - env = os.environ.get("HERMES_KANBAN_STOP_NUDGE") - if env is not None and env.strip().lower() in {"0", "false", "no", "off"}: + if (os.environ.get("HERMES_KANBAN_STOP_NUDGE") or "").strip().lower() in {"0", "false", "no", "off"}: return False return bool((os.environ.get("HERMES_KANBAN_TASK") or "").strip()) def _tool_call_name(tc: Any) -> str: + """Tool name from a dict or object tool call (``function.name`` first, then ``name``).""" if isinstance(tc, dict): fn = tc.get("function") - if isinstance(fn, dict): - return str(fn.get("name") or "") - return str(tc.get("name") or "") + return str((fn.get("name") if isinstance(fn, dict) else tc.get("name")) or "") fn = getattr(tc, "function", None) - if fn is not None: - return str(getattr(fn, "name", "") or "") - return str(getattr(tc, "name", "") or "") + return str((getattr(fn, "name", "") if fn is not None else getattr(tc, "name", "")) or "") def session_called_kanban_terminal(messages: Iterable[dict] | None) -> bool: """True if this conversation already invoked a terminal kanban tool.""" - if not messages: - return False - for msg in messages: - if not isinstance(msg, dict): - continue + for msg in filter(lambda m: isinstance(m, dict), messages or ()): role = msg.get("role") - if role == "assistant": - if any(_tool_call_name(tc) in _TERMINAL_KANBAN_TOOLS for tc in msg.get("tool_calls") or []): - return True - elif role == "tool" and str(msg.get("name") or "") in _TERMINAL_KANBAN_TOOLS: + if role == "assistant" and any( + _tool_call_name(tc) in _TERMINAL_KANBAN_TOOLS for tc in msg.get("tool_calls") or [] + ): + return True + if role == "tool" and str(msg.get("name") or "") in _TERMINAL_KANBAN_TOOLS: return True return False @@ -60,11 +51,8 @@ def build_kanban_stop_nudge( max_attempts: int = _DEFAULT_MAX_ATTEMPTS, task_id: Optional[str] = None, ) -> Optional[str]: - """Synthetic follow-up when a kanban worker exits without a terminal tool. - - ``None`` when the guard should not fire (not a kanban worker, already - completed/blocked, or nudge budget exhausted). - """ + """Synthetic follow-up when a kanban worker exits without a terminal tool; ``None`` when + the guard should not fire (not a kanban worker, already completed/blocked, budget exhausted).""" if ( not kanban_stop_nudge_enabled() or attempts >= max_attempts @@ -88,8 +76,4 @@ def build_kanban_stop_nudge( ) -__all__ = [ - "build_kanban_stop_nudge", - "kanban_stop_nudge_enabled", - "session_called_kanban_terminal", -] +__all__ = ["build_kanban_stop_nudge", "kanban_stop_nudge_enabled", "session_called_kanban_terminal"] diff --git a/tests/agent/test_file_safety_cross_profile.py b/tests/agent/test_file_safety_cross_profile.py index 3be8ea8791..8be6481775 100644 --- a/tests/agent/test_file_safety_cross_profile.py +++ b/tests/agent/test_file_safety_cross_profile.py @@ -1,4 +1,4 @@ -"""Tests for the (retired) cross-Hermes-profile write guard in agent/file_safety.""" +"""Tests for the active-profile resolver in agent/file_safety.""" from __future__ import annotations from pathlib import Path @@ -88,26 +88,3 @@ class TestResolveActiveProfileName: monkeypatch.setattr(fs, "_hermes_home_path", _boom) # Should not raise — falls back to "default" assert fs._resolve_active_profile_name() == "default" - - -# --------------------------------------------------------------------------- -# get_cross_profile_warning -# --------------------------------------------------------------------------- - - -class TestGetCrossProfileWarning: - """The guard is RETIRED (maintainer decision): profiles are not - isolated, so the warning helper is a permanent None stub — kept only - so external callers fail soft. The classifier itself survives for - the system-prompt hint and diagnostics.""" - - def test_in_profile_returns_none(self, fake_hermes, monkeypatch): - from agent.file_safety import get_cross_profile_warning - assert get_cross_profile_warning( - str(fake_hermes["root"] / "skills" / "a" / "SKILL.md")) is None - - def test_cross_profile_returns_none_guard_retired(self, fake_hermes, monkeypatch): - from agent.file_safety import get_cross_profile_warning - target = fake_hermes["root"] / "profiles" / "security" / "skills" / "x" / "SKILL.md" - assert get_cross_profile_warning(str(target)) is None - diff --git a/tests/agent/test_save_url_image.py b/tests/agent/test_save_url_image.py index 3737871f67..086182331a 100644 --- a/tests/agent/test_save_url_image.py +++ b/tests/agent/test_save_url_image.py @@ -123,9 +123,10 @@ class TestSaveUrlImage: def test_oversize_raises_and_cleans_up(self, http_server, tmp_path): """Oversize downloads must NOT leak a partial file into the cache.""" base, _ = http_server - from agent.image_gen_provider import save_url_image, _images_cache_dir + from agent import provider_media + from agent.image_gen_provider import save_url_image - cache_dir = _images_cache_dir() + cache_dir = provider_media.cache_dir("images") before = set(cache_dir.glob("*")) with pytest.raises(ValueError, match="exceeds"): save_url_image(f"{base}/oversize", max_bytes=1024 * 1024)