refactor(agent/image+safety): route vision probes/inline executors on data tables, dedupe path guards (-25% LOC)
- image_routing: capability probes -> _VISION_PROBES ordered table; magic-byte sniffing -> _MAGIC signature table; bool tokens -> dict; shared code-span / file-existence helpers for extract_image_refs. Fixture-corpus diff base vs new identical. - inline_tool_executors: 11 hand-written executor bodies -> _tool/_callback_tool arg-spec factories (kwargs mapping unchanged; corpus of calls identical). - file_safety: unify _is_under/_under_any, _resolve_target/_resolve_each, _mirror_info; deny-list tables compacted; drop retired get_cross_profile_warning stub (+ its tests). Every rule preserved (corpus). - image_gen_provider: drop _images_cache_dir (test-only wrapper), compact. - estop/delegation_context/kanban_stop/image_gen_registry: defensive-layer collapse and docstring compaction. Tool schema dump byte-identical to base.
This commit is contained in:
Teknium
@@ -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)
|
||||
|
||||
@@ -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/<n>) 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
|
||||
|
||||
@@ -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 <root>/... stay guarded when
|
||||
running under a profile (HERMES_HOME = <root>/profiles/<name>).
|
||||
"""
|
||||
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 / <root>. 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 / <root>: (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:
|
||||
# <HERMES_HOME>/profiles/<name>/sandboxes/<backend>/<task>/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 / <backend> / <task> / home / .hermes / <thing>; 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),
|
||||
|
||||
@@ -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
|
||||
``<repo>/plugins/image_gen/<name>/`` (built-in) or
|
||||
``~/.hermes/plugins/image_gen/<name>/`` (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
|
||||
``<repo>/plugins/image_gen/<name>/`` or ``~/.hermes/plugins/image_gen/<name>/``);
|
||||
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,
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"(?<![/:\w.])(?:~/|/)(?:[\w.\-]+/)*[\w.\-]+\.(?:" + _IMAGE_EXT_PATTERN + r")\b",
|
||||
re.IGNORECASE,
|
||||
r"(?<![/:\w.])(?:~/|/)(?:[\w.\-]+/)*[\w.\-]+\.(?:" + _IMAGE_EXT_PATTERN + r")\b", re.IGNORECASE,
|
||||
)
|
||||
|
||||
# http(s) URL ending in an image extension, optional query string. Strict
|
||||
# ``http(s)://`` so ``file://`` and other schemes are not grabbed.
|
||||
_IMAGE_URL_RE = re.compile(
|
||||
r"https?://[^\s<>\"']+?\.(?:" + _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.<key>`` → ``providers.<name>.<key>`` → ``custom_providers[].<key>``,
|
||||
where ``<name>`` 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.<key>`` → ``providers.<name>.<key>``
|
||||
→ ``custom_providers[].<key>``, ``<name>`` 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 <svg tag near the start (skip BOM/whitespace).
|
||||
head = raw[:512].lstrip().lower()
|
||||
if (head.startswith(b"<?xml") or head.startswith(b"<svg")) and b"<svg" in head:
|
||||
return "image/svg+xml"
|
||||
return None
|
||||
return "image/svg+xml" if head.startswith((b"<?xml", b"<svg")) and b"<svg" in head else None
|
||||
|
||||
|
||||
# Formats every major vision provider accepts natively. Anything else must be
|
||||
# transcoded to PNG before declaring media_type or the provider returns HTTP
|
||||
# 400 and the whole turn fails. Chat platforms freely accept AVIF (Chromium
|
||||
# screenshots), HEIC (iPhone), TIFF, BMP and ICO, so users do hit this. SVG is
|
||||
# vector — Pillow cannot rasterize it — and is skipped (logged) instead.
|
||||
_UNIVERSALLY_SUPPORTED_MIMES = frozenset({
|
||||
"image/png", "image/jpeg", "image/gif", "image/webp",
|
||||
})
|
||||
# Formats every major vision provider accepts natively. Anything else is transcoded
|
||||
# to PNG before declaring media_type or the provider returns HTTP 400 and the turn
|
||||
# fails; chat platforms freely accept AVIF (Chromium screenshots), HEIC (iPhone),
|
||||
# TIFF, BMP and ICO. SVG is vector — Pillow cannot rasterize it — so it is skipped.
|
||||
_UNIVERSALLY_SUPPORTED_MIMES = frozenset({"image/png", "image/jpeg", "image/gif", "image/webp"})
|
||||
|
||||
|
||||
def _transcode_to_png(raw: bytes) -> 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: <path>]`` / ``[Image attached: <url>]`` — 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: <path>]`` / ``[Image attached: <url>]``
|
||||
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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user