refactor(agent/misc-g6): dedupe qwen prep, image-part checks, pricing normalizers; tighten provider ABC/registry modules

agent/tts_provider.py

@@ -1,23 +1,11 @@
-"""
-Text-to-Speech Provider ABC
-============================
+"""Text-to-Speech Provider ABC.
 
-Pluggable-backend interface for TTS synthesis. Providers register via
-``PluginContext.register_tts_provider()``; the one named by ``tts.provider``
-services ``text_to_speech`` **only when that name is neither a built-in nor a
-``tts.providers.<name>: type: command`` entry**. Resolution order:
-
-1. Built-in providers (``BUILTIN_TTS_PROVIDERS`` in :mod:`tools.tts_tool`) —
-   always win; :func:`agent.tts_registry.register_provider` rejects colliding
-   names and the dispatcher re-checks at dispatch time.
-2. Command-type providers from ``config.yaml`` — win over a same-name plugin
-   because config is more local than a plugin install.
-3. Plugin providers (this ABC) — for backends needing a Python SDK, streaming
-   bytes, OAuth refresh, or voice-listing APIs the shell template can't express.
-
-:meth:`TTSProvider.synthesize` writes audio to ``output_path`` and returns the
-path; it should raise on failure — the dispatcher converts exceptions into the
-standard ``{success: False, error: …}`` envelope.
+Providers register via ``PluginContext.register_tts_provider()`` and service
+``text_to_speech`` only when ``tts.provider`` names neither a built-in
+(``BUILTIN_TTS_PROVIDERS`` in :mod:`tools.tts_tool`; the registry rejects
+colliding names) nor a ``tts.providers.<name>: type: command`` entry (config is
+more local than a plugin install, so it wins). :meth:`TTSProvider.synthesize`
+should raise on failure — the dispatcher builds the ``{success: False}`` envelope.
 """
 
 from __future__ import annotations
@@ -50,9 +38,7 @@ class TTSProvider(CatalogProviderBase):
     def default_voice(self) -> Optional[str]:
         """Id of the first voice entry, or None if not applicable."""
         voices = self.list_voices()
-        if voices:
-            return voices[0].get("id")
-        return None
+        return voices[0].get("id") if voices else None
 
     @abc.abstractmethod
     def synthesize(
@@ -88,10 +74,8 @@ class TTSProvider(CatalogProviderBase):
 
     def warm(self) -> None:
         """Speech output was just turned on; pre-load so the first reply is hot.
-
-        Called from the TTS lease path (Desktop read-aloud / voice conversation)
-        when this is the configured provider. Best-effort; default no-op.
-        """
+        Called from the TTS lease path when this is the configured provider.
+        Best-effort; default no-op."""
 
     def release(self) -> None:
         """Last speech-output lease released; free resident resources (counterpart
@@ -100,11 +84,9 @@ class TTSProvider(CatalogProviderBase):
     @property
     def voice_compatible(self) -> bool:
         """Whether output suits voice-bubble delivery (mirrors
-        ``tts.providers.<name>.voice_compatible``).
-
-        True → the gateway converts to Opus via ffmpeg if needed; False →
-        delivered as a regular audio attachment. Default False (opt in).
-        """
+        ``tts.providers.<name>.voice_compatible``): True → the gateway converts
+        to Opus via ffmpeg if needed; False → regular audio attachment. Default
+        False (opt in)."""
         return False
 
 
@@ -112,9 +94,5 @@ def resolve_output_format(value: Optional[str]) -> str:
     """Clamp an output_format to :data:`VALID_OUTPUT_FORMATS`; invalid values
     coerce to :data:`DEFAULT_OUTPUT_FORMAT` so the tool surface forgives agent
     mistakes instead of rejecting them."""
-    if not isinstance(value, str):
-        return DEFAULT_OUTPUT_FORMAT
-    v = value.strip().lower()
-    if v in VALID_OUTPUT_FORMATS:
-        return v
-    return DEFAULT_OUTPUT_FORMAT
+    v = value.strip().lower() if isinstance(value, str) else None
+    return v if v in VALID_OUTPUT_FORMATS else DEFAULT_OUTPUT_FORMAT

agent/tts_registry.py

@@ -1,21 +1,11 @@
-"""
-TTS Provider Registry
-=====================
+"""TTS Provider Registry.
 
-Central map of registered TTS providers. Populated by plugins at import-time
-via :meth:`PluginContext.register_tts_provider`; consumed by
-:mod:`tools.tts_tool` to dispatch ``text_to_speech`` calls to the active
-plugin backend **when** the configured ``tts.provider`` name is neither a
-built-in nor a command-type provider.
-
-Built-ins-always-win: a plugin name colliding with a built-in TTS provider is
-rejected at registration with a warning (re-checked at dispatch time in
-:func:`tools.tts_tool._dispatch_to_plugin_provider`).
-
-Command-providers-win-over-plugins is enforced by the dispatcher, not here:
-it checks for a same-name ``tts.providers.<name>: type: command`` entry before
-consulting the registry (a name declared in the user's config.yaml is more
-specific to their setup than an installed plugin).
+Registered plugin TTS providers, populated at import-time via
+:meth:`PluginContext.register_tts_provider` and consulted by :mod:`tools.tts_tool`
+only when ``tts.provider`` is neither a built-in nor a command-type provider.
+Built-ins always win: a colliding plugin name is rejected here with a warning
+(re-checked at dispatch). Command-providers-win-over-plugins is enforced by the
+dispatcher (a name in the user's config.yaml is more specific than a plugin).
 """
 
 from __future__ import annotations
@@ -29,21 +19,12 @@ logger = logging.getLogger(__name__)
 
 
 # Names reserved for native built-in TTS handlers. **Kept in sync with
-# ``BUILTIN_TTS_PROVIDERS`` in :mod:`tools.tts_tool`** (a regression test in
-# ``tests/agent/test_tts_registry.py::TestBuiltinSync`` fails on drift);
-# importing it directly would be a circular import.
+# ``BUILTIN_TTS_PROVIDERS`` in :mod:`tools.tts_tool`** (``TestBuiltinSync`` in
+# ``tests/agent/test_tts_registry.py`` fails on drift); importing it directly
+# would be a circular import.
 _BUILTIN_NAMES = frozenset({
-    "edge",
-    "elevenlabs",
-    "openai",
-    "minimax",
-    "xai",
-    "mistral",
-    "gemini",
-    "neutts",
-    "kittentts",
-    "piper",
-    "deepinfra",
+    "edge", "elevenlabs", "openai", "minimax", "xai", "mistral", "gemini", "neutts", "kittentts",
+    "piper", "deepinfra",
 })
 
 

agent/usage_pricing.py

@@ -391,33 +391,34 @@ def resolve_billing_route(
     return BillingRoute(provider=provider_name or "unknown", model=bare if model else "", base_url=url, billing_mode="unknown")
 
 
-def _normalize_bedrock_model_name(model: str) -> str:
-    """Normalize a Bedrock model id to its bare foundation-model form.
+_BEDROCK_REGION_PREFIXES = ("global.", "us.", "eu.", "apac.", "ap.", "au.", "jp.", "ca.", "sa.", "me.", "af.")
+# Bedrock ids end in documented date/revision/profile components (``-20250514-v1:0``).
+_BEDROCK_TRAILERS = (r":\d+$", r"-v\d+$", r"-\d{8}$")
 
-    Cross-region inference profiles prefix the id with a region scope
-    (``us.``/``global.``/``apac.``/``au.``/...); the pricing table is keyed on
-    the bare ``anthropic.claude-*`` id, so the prefix is stripped. Also maps
-    dotted versions (``4.7`` → ``4-7``) and strips only the documented
-    trailing date/revision/profile components (``-20250514-v1:0``).
-    """
-    name = model.lower().strip()
-    for prefix in ("global.", "us.", "eu.", "apac.", "ap.", "au.", "jp.", "ca.", "sa.", "me.", "af."):
-        if name.startswith(prefix):
-            name = name[len(prefix):]
-            break
-    name = re.sub(r"(\d+)\.(\d+)", r"\1-\2", name)
-    name = re.sub(r":\d+$", "", name)
-    name = re.sub(r"-v\d+$", "", name)
-    name = re.sub(r"-\d{8}$", "", name)
+
+def _strip_prefix(name: str, prefixes: tuple[str, ...]) -> str:
+    """Drop the first matching prefix (at most one), else return ``name`` unchanged."""
+    return next((name[len(p):] for p in prefixes if name.startswith(p)), name)
+
+
+def _normalize_bedrock_model_name(model: str) -> str:
+    """Bare foundation-model id: strip the cross-region inference-profile scope
+    (``us.``/``global.``/...), map dotted versions (``4.7`` → ``4-7``), then
+    strip the trailing date/revision/profile components."""
+    name = re.sub(r"(\d+)\.(\d+)", r"\1-\2", _strip_prefix(model.lower().strip(), _BEDROCK_REGION_PREFIXES))
+    for pattern in _BEDROCK_TRAILERS:
+        name = re.sub(pattern, "", name)
     return name
 
 
 def _normalize_anthropic_model_name(model: str) -> str:
     """Strip an ``anthropic/`` prefix and map dotted versions (4.7 → 4-7)."""
-    name = model.lower().strip()
-    if name.startswith("anthropic/"):
-        name = name[len("anthropic/"):]
-    return re.sub(r"(\d+)\.(\d+)", r"\1-\2", name)
+    return re.sub(r"(\d+)\.(\d+)", r"\1-\2", _strip_prefix(model.lower().strip(), ("anthropic/",)))
+
+
+# Anthropic dot-notation (opus-4.7) and Bedrock region-prefixed ids need
+# normalizing before a second lookup.
+_MODEL_NORMALIZERS = {"anthropic": _normalize_anthropic_model_name, "bedrock": _normalize_bedrock_model_name}
 
 
 def _lookup_official_docs_pricing(route: BillingRoute) -> Optional[PricingEntry]:
@@ -425,18 +426,9 @@ def _lookup_official_docs_pricing(route: BillingRoute) -> Optional[PricingEntry]
     entry = _OFFICIAL_DOCS_PRICING.get((route.provider, model))
     if entry:
         return entry
-    # Anthropic dot-notation (opus-4.7) and Bedrock region-prefixed ids need
-    # normalizing before a second lookup.
-    normalize = {
-        "anthropic": _normalize_anthropic_model_name, "bedrock": _normalize_bedrock_model_name
-    }.get(route.provider)
-    if normalize:
-        normalized = normalize(model)
-        if normalized != model:
-            entry = _OFFICIAL_DOCS_PRICING.get((route.provider, normalized))
-            if entry:
-                return entry
-    return None
+    normalize = _MODEL_NORMALIZERS.get(route.provider)
+    normalized = normalize(model) if normalize else model
+    return _OFFICIAL_DOCS_PRICING.get((route.provider, normalized)) if normalized != model else None
 
 
 def _openrouter_pricing_entry(route: BillingRoute) -> Optional[PricingEntry]:
@@ -453,30 +445,25 @@ def _pricing_entry_from_metadata(
     if model_id not in metadata:
         return None
     pricing = metadata[model_id].get("pricing") or {}
-    prompt = _to_decimal(pricing.get("prompt"))
-    completion = _to_decimal(pricing.get("completion"))
-    request = _to_decimal(pricing.get("request"))
-    cache_read = _to_decimal(
-        pricing.get("cache_read") or pricing.get("cached_prompt") or pricing.get("input_cache_read")
-    )
-    cache_write = _to_decimal(
-        pricing.get("cache_write")
-        or pricing.get("cache_creation")
-        or pricing.get("input_cache_write")
-    )
-    if prompt is None and completion is None and request is None:
-        return None
 
-    def _per_million(value: Optional[Decimal]) -> Optional[Decimal]:
+    def per_million(key: str, *aliases: str) -> Optional[Decimal]:
+        raw = pricing.get(key)
+        for alias in aliases:  # alias chain is truthiness-based (``a or b or c``)
+            raw = raw or pricing.get(alias)
+        value = _to_decimal(raw)
         return None if value is None else value * _ONE_MILLION
 
+    prompt = per_million("prompt")
+    completion = per_million("completion")
+    request = _to_decimal(pricing.get("request"))
+    if prompt is None and completion is None and request is None:
+        return None
     return PricingEntry(
-        input_cost_per_million=_per_million(prompt),
-        output_cost_per_million=_per_million(completion),
-        cache_read_cost_per_million=_per_million(cache_read),
-        cache_write_cost_per_million=_per_million(cache_write), request_cost=request,
-        source="provider_models_api", source_url=source_url, pricing_version=pricing_version,
-        fetched_at=_UTC_NOW(),
+        input_cost_per_million=prompt, output_cost_per_million=completion,
+        cache_read_cost_per_million=per_million("cache_read", "cached_prompt", "input_cache_read"),
+        cache_write_cost_per_million=per_million("cache_write", "cache_creation", "input_cache_write"),
+        request_cost=request, source="provider_models_api", source_url=source_url,
+        pricing_version=pricing_version, fetched_at=_UTC_NOW(),
     )
 
 
@@ -609,32 +596,24 @@ def estimate_usage_cost(
 
     # Whole-request context tier (e.g. Gemini Pro >200k prompts): above the
     # threshold the *_above rates apply to the entire request; None falls back.
-    input_rate = entry.input_cost_per_million
-    output_rate = entry.output_cost_per_million
-    cache_read_rate = entry.cache_read_cost_per_million
-    cache_write_rate = entry.cache_write_cost_per_million
-    if entry.tier_threshold_tokens is not None and usage.prompt_tokens > entry.tier_threshold_tokens:
-        if entry.input_cost_per_million_above is not None:
-            input_rate = entry.input_cost_per_million_above
-        if entry.output_cost_per_million_above is not None:
-            output_rate = entry.output_cost_per_million_above
-        if entry.cache_read_cost_per_million_above is not None:
-            cache_read_rate = entry.cache_read_cost_per_million_above
-
-    if usage.input_tokens and input_rate is None:
-        return _unknown_cost(entry.source)
-    if usage.output_tokens and output_rate is None:
-        return _unknown_cost(entry.source)
-    if usage.cache_read_tokens and cache_read_rate is None:
-        return _unknown_cost(entry.source, "cache-read pricing unavailable for route")
-    if usage.cache_write_tokens and cache_write_rate is None:
-        return _unknown_cost(entry.source, "cache-write pricing unavailable for route")
+    above = entry.tier_threshold_tokens is not None and usage.prompt_tokens > entry.tier_threshold_tokens
+    buckets = []
+    for tokens, rate, rate_above, note in (
+        (usage.input_tokens, entry.input_cost_per_million, entry.input_cost_per_million_above, ()),
+        (usage.output_tokens, entry.output_cost_per_million, entry.output_cost_per_million_above, ()),
+        (usage.cache_read_tokens, entry.cache_read_cost_per_million, entry.cache_read_cost_per_million_above,
+         ("cache-read pricing unavailable for route",)),
+        (usage.cache_write_tokens, entry.cache_write_cost_per_million, None,
+         ("cache-write pricing unavailable for route",)),
+    ):
+        if above and rate_above is not None:
+            rate = rate_above
+        if tokens and rate is None:
+            return _unknown_cost(entry.source, *note)
+        buckets.append((tokens, rate))
 
     amount = _ZERO
-    for tokens, rate in (
-        (usage.input_tokens, input_rate), (usage.output_tokens, output_rate),
-        (usage.cache_read_tokens, cache_read_rate), (usage.cache_write_tokens, cache_write_rate),
-    ):
+    for tokens, rate in buckets:
         if rate is not None:
             amount += Decimal(tokens) * rate / _ONE_MILLION
     if entry.request_cost is not None and usage.request_count:

agent/verification_evidence.py

@@ -8,6 +8,7 @@ and never upgrades targeted checks into "repo green".
 from __future__ import annotations
 
 import json
+import re
 import shlex
 import sqlite3
 import tempfile
@@ -135,9 +136,8 @@ def _connect() -> sqlite3.Connection:
 def _transaction() -> Iterator[sqlite3.Connection]:
     """Open a connection, commit/rollback on exit, and ALWAYS close it.
 
-    ``sqlite3.Connection`` as a context manager only commits/rolls back; it does
-    not close. Relying on it alone leaks a connection (and its WAL/SHM fds) per
-    call until GC runs, which can exhaust ``RLIMIT_NOFILE`` in a long process.
+    ``sqlite3.Connection`` as a context manager only commits/rolls back; without
+    the close, each call leaks a connection (and WAL/SHM fds) until GC runs.
     """
     conn = _connect()
     try:
@@ -249,28 +249,20 @@ def _exit_status_is_attributable(segments: list[_ShellSegment], match_index: int
 
 def _canonical_tokens(canonical: str) -> list[str]:
     """Tokenize a canonical command, stripping leading ``./`` from each token."""
-    def clean(token: str) -> str:
-        token = token.strip()
-        while token.startswith("./"):
-            token = token[2:]
-        return token
-
     try:
-        return [clean(t) for t in shlex.split(canonical) if t]
+        return [re.sub(r"^(?:\./)+", "", t.strip()) for t in shlex.split(canonical) if t]
     except ValueError:
         return []
 
 
 def _strip_command_prefix(tokens: list[str]) -> list[str]:
     """Remove harmless command prefixes (env, VAR=x, command/time/noglob)."""
-    remaining = list(tokens)
-    if remaining and remaining[0] == "env":
-        remaining = remaining[1:]
-    while remaining and "=" in remaining[0] and not remaining[0].startswith("-"):
-        remaining = remaining[1:]
-    while remaining and remaining[0] in {"command", "time", "noglob"}:
-        remaining = remaining[1:]
-    return remaining
+    i = 1 if tokens and tokens[0] == "env" else 0
+    while i < len(tokens) and "=" in tokens[i] and not tokens[i].startswith("-"):
+        i += 1
+    while i < len(tokens) and tokens[i] in {"command", "time", "noglob"}:
+        i += 1
+    return list(tokens[i:])
 
 
 def _equivalent_needles(needle: list[str]) -> list[list[str]]:
@@ -452,12 +444,12 @@ def classify_verification_command(
     verify_commands = list(facts.get("verifyCommands") or [])
     match = _find_canonical_match(command, verify_commands, int(exit_code))
     is_ad_hoc = False
-    if match is None and not verify_commands:
-        ad_hoc_args = _find_ad_hoc_match(command, facts.get("root"), int(exit_code))
-        is_ad_hoc = ad_hoc_args is not None
-        match = ("ad-hoc verification script", ad_hoc_args) if is_ad_hoc else None
     if match is None:
-        return None
+        ad_hoc_args = None if verify_commands else _find_ad_hoc_match(command, facts.get("root"), int(exit_code))
+        if ad_hoc_args is None:
+            return None
+        is_ad_hoc = True
+        match = ("ad-hoc verification script", ad_hoc_args)
 
     canonical, trailing_args = match
     return VerificationEvidence(
@@ -484,10 +476,9 @@ def record_verify_run(
 ) -> Optional[dict[str, Any]]:
     """Record a completed ``hermes verify`` run as verification evidence.
 
-    Explicit CLI-side write with nothing to classify: a pass marks the workspace
-    ``passed`` for the verify-on-stop guard like a canonical test command would;
-    a failure keeps the guard asking for a fix. ``root`` is re-resolved through
-    project facts so it matches what :func:`verification_status` derives later.
+    A pass marks the workspace ``passed`` for the verify-on-stop guard like a
+    canonical test command would. ``root`` is re-resolved through project facts
+    so it matches what :func:`verification_status` derives later.
     """
     resolved = str(Path(root).resolve())
     return _insert_evidence(VerificationEvidence(
@@ -596,5 +587,4 @@ def verification_status(*, session_id: str | None, cwd: str | Path | None) -> di
 
     evidence = dict(event)
     stale = bool(state["last_edit_at"]) and state["last_edit_at"] > evidence["created_at"]
-    result["evidence"] = evidence
-    return {"status": "stale" if stale else evidence["status"], **result}
+    return {"status": "stale" if stale else evidence["status"], **result, "evidence": evidence}

agent/verification_stop.py

@@ -42,9 +42,8 @@ def _is_non_code_path(raw: str) -> bool:
 
 
 def _session_is_messaging_surface() -> bool:
-    """Whether this turn is delivered over a human messaging channel
-    (``gateway.session_context``). An unreachable gateway package means no
-    messaging channel, so report a local surface (keeps verify-on-stop enabled)."""
+    """Whether this turn is delivered over a human messaging channel. An
+    unreachable gateway package means no messaging channel (verify-on-stop stays on)."""
     try:
         from gateway.session_context import session_is_messaging_surface
 
@@ -56,12 +55,11 @@ def _session_is_messaging_surface() -> bool:
 def verify_on_stop_enabled(config: dict[str, Any] | None = None) -> bool:
     """Return whether edit -> verify-before-finish behavior is enabled.
 
-    Precedence: explicit ``HERMES_VERIFY_ON_STOP`` env var, then explicit
-    ``agent.verify_on_stop`` config. Default OFF (opt-in). A bool forces the
-    behavior; ``"auto"`` is the legacy surface-aware mode: ON for interactive
-    coding surfaces (CLI, TUI, desktop) and programmatic callers, OFF for
-    messaging surfaces where the verification narrative is chat noise.
-    Missing/unrecognized values fall back to OFF.
+    Precedence: ``HERMES_VERIFY_ON_STOP`` env var, then ``agent.verify_on_stop``
+    config; default OFF (opt-in). A bool forces the behavior; ``"auto"`` is the
+    legacy surface-aware mode: ON for interactive coding surfaces and
+    programmatic callers, OFF for messaging surfaces where the verification
+    narrative is chat noise. Missing/unrecognized values fall back to OFF.
     """
     env = os.environ.get("HERMES_VERIFY_ON_STOP")
     if env is not None:
@@ -77,13 +75,10 @@ def verify_on_stop_enabled(config: dict[str, Any] | None = None) -> bool:
     cfg_val = agent_cfg.get("verify_on_stop") if isinstance(agent_cfg, dict) else None
     if isinstance(cfg_val, bool):
         return cfg_val
-    if isinstance(cfg_val, str):
-        token = cfg_val.strip().lower()
-        if token == "auto":
-            return not _session_is_messaging_surface()
-        if token in _TRUTHY_TOKENS | _FALSY_TOKENS:
-            return token in _TRUTHY_TOKENS
-    return False
+    token = cfg_val.strip().lower() if isinstance(cfg_val, str) else ""
+    if token == "auto":
+        return not _session_is_messaging_surface()
+    return token in _TRUTHY_TOKENS
 
 
 def _candidate_cwds(paths: Iterable[str]) -> list[Path]:
@@ -139,13 +134,12 @@ def _workspace_has_runnable_recipe(root: Any) -> bool:
     if not root:
         return False
     try:
-        root_path = Path(str(root))
         from agent.verify.environment import manifest_path
-
-        if manifest_path(root_path).is_file():
-            return True
         from agent.verify.recipes import detect_recipe
 
+        root_path = Path(str(root))
+        if manifest_path(root_path).is_file():
+            return True
         recipe = detect_recipe(root_path)
         return bool(recipe is not None and recipe.start)
     except Exception:

agent/verify/recipes.py

@@ -1,17 +1,12 @@
 """Static run-recipe detection for project verification.
 
-Ported nearly 1:1 from superagent-ai/grok-cli ``src/verify/recipes.ts``, keeping
-its detection order and command choices: Node (lockfile-based package manager +
-framework), Python (Django / FastAPI / Flask / generic, uv/poetry/pipenv aware),
-Go, Rust, Java (Maven/Gradle), Makefile fallback, docker-compose.
-
-Layer ownership vs :mod:`agent.coding_context`: ``detect_project_facts`` owns the
-cheap, byte-stable prompt-time facts (manifests, package managers, verify
-commands) — never push runtime detection into it. This module owns the deep
-runtime recipe (framework id, bootstrap/build/test, start command, port,
-readiness path) that lets ``hermes verify`` boot the app and prove it serves
-HTTP; the CLI merges project-facts verify commands the recipe missed into its
-test list (``hermes_cli.verify_cmd._merge_project_facts_commands``).
+Ported nearly 1:1 from superagent-ai/grok-cli ``src/verify/recipes.ts`` (same
+detection order and command choices). Layer ownership: ``detect_project_facts``
+in :mod:`agent.coding_context` owns the cheap, byte-stable prompt-time facts —
+never push runtime detection into it. This module owns the deep runtime recipe
+(framework, bootstrap/build/test, start command, port, readiness path) that lets
+``hermes verify`` boot the app; the CLI merges project-facts verify commands the
+recipe missed (``hermes_cli.verify_cmd._merge_project_facts_commands``).
 """
 
 from __future__ import annotations
@@ -24,11 +19,8 @@ from typing import Any
 
 
 def _as_strings(value: Any) -> list[str]:
-    if isinstance(value, str) and value.strip():
-        return [value.strip()]
-    if not isinstance(value, list):
-        return []
-    return [v.strip() for v in value if isinstance(v, str) and v.strip()]
+    values = [value] if isinstance(value, str) else value if isinstance(value, list) else []
+    return [v.strip() for v in values if isinstance(v, str) and v.strip()]
 
 
 @dataclass
@@ -66,7 +58,7 @@ class Recipe:
         name = raw.get("name") or raw.get("appLabel")
         if not isinstance(name, str) or not name.strip():
             return None
-        kind = raw.get("kind") or raw.get("appKind") or "unknown"
+        kind = raw.get("kind") or raw.get("appKind")
         if not isinstance(kind, str) or not kind.strip():
             kind = "unknown"
 
@@ -99,12 +91,9 @@ def _read_text(root: Path, name: str) -> str | None:
 
 
 def _read_package_json(root: Path) -> dict[str, Any] | None:
-    raw = _read_text(root, "package.json")
-    if raw is None:
-        return None
     try:
-        parsed = json.loads(raw)
-    except (json.JSONDecodeError, ValueError):
+        parsed = json.loads(_read_text(root, "package.json") or "")
+    except ValueError:  # includes JSONDecodeError; also the missing-file "" case
         return None
     return parsed if isinstance(parsed, dict) else None
 
@@ -159,8 +148,7 @@ def _detect_node_recipe(root: Path, pkg: dict[str, Any]) -> Recipe:
     raw_scripts = pkg.get("scripts")
     scripts: dict[str, str] = raw_scripts if isinstance(raw_scripts, dict) else {}
     deps: dict[str, Any] = {}
-    for key in ("dependencies", "devDependencies"):
-        section = pkg.get(key)
+    for section in (pkg.get("dependencies"), pkg.get("devDependencies")):
         if isinstance(section, dict):
             deps.update(section)
 
@@ -288,14 +276,12 @@ def _detect_make_recipe(root: Path) -> Recipe | None:
         return None
     targets = [m.group(1) for m in map(_MAKE_TARGET_RE.match, makefile.splitlines()) if m]
     picked = {
-        phase: next((f"make {n}" for n in names if n in targets), None)
+        phase: [f"make {n}" for n in names if n in targets][:1]
         for phase, names in _MAKE_PHASE_TARGETS.items()
     }
     return Recipe(
-        name="Makefile-driven project", kind="make", start=picked["start"],
-        bootstrap=[picked["bootstrap"]] if picked["bootstrap"] else [],
-        build=[picked["build"]] if picked["build"] else [],
-        test=[picked["test"]] if picked["test"] else [],
+        name="Makefile-driven project", kind="make", start=(picked["start"] or [None])[0],
+        bootstrap=picked["bootstrap"], build=picked["build"], test=picked["test"],
         evidence=["Detected Makefile", f"Targets: {', '.join(targets) or '(none)'}"],
     )
 

agent/verify/runner.py

@@ -149,8 +149,8 @@ def _terminate_process_group(proc: subprocess.Popen) -> None:
             pgid = None
 
     def stop(sig: int, fallback: Callable[[], None]) -> None:
-        if pgid is not None and killpg is not None:
-            killpg(pgid, sig)  # windows-footgun: ok — POSIX-only branch (killpg checked above)
+        if pgid is not None:
+            killpg(pgid, sig)  # windows-footgun: ok — POSIX-only branch (pgid only set when killpg exists)
         else:
             fallback()
 
@@ -186,8 +186,7 @@ def _run_start_phase(
     finally:
         _terminate_process_group(proc)
         try:
-            if proc.stdout is not None:
-                output = proc.stdout.read() or ""
+            output = proc.stdout.read() or "" if proc.stdout is not None else ""
         except (OSError, ValueError):
             output = ""
     return ReadinessResult(url, ready, status, time.monotonic() - started, error, _tail(output))
@@ -218,9 +217,6 @@ def run_verify(
             if not phase_result.ok and stop_on_failure:
                 return result
 
-    failed = not all(p.ok for p in result.phases)
-    if skip_start or "start" not in selected or failed or not recipe.start:
-        return result
-
-    result.readiness = _run_start_phase(recipe, root, ready_timeout, port_override)
+    if not skip_start and "start" in selected and recipe.start and all(p.ok for p in result.phases):
+        result.readiness = _run_start_phase(recipe, root, ready_timeout, port_override)
     return result

agent/verify_hooks.py

@@ -1,9 +1,8 @@
 """Verification-loop helpers for the ``pre_verify`` round-end gate.
 
-After code edits, the loop fires ``pre_verify`` (directives resolved by
-:func:`hermes_cli.plugins.get_pre_verify_continue_message`); a directive keeps the
-agent going one more turn. The shipped coding guidance rides on the evidence-based
-verification-stop nudge (``agent/verification_stop.py``) rather than a second
+After code edits the loop fires ``pre_verify`` (directives resolved by
+:func:`hermes_cli.plugins.get_pre_verify_continue_message`). The shipped coding
+guidance rides on the evidence-based verification-stop nudge rather than a second
 default stop gate, so default token cost stays tied to the "missing verification
 evidence" decision while ``pre_verify`` remains free for user/plugin policy.
 """

agent/video_gen_provider.py

@@ -1,38 +1,24 @@
-"""
-Video Generation Provider ABC
-=============================
+"""Video Generation Provider ABC.
 
-Pluggable-backend interface for video generation. Providers register via
-``PluginContext.register_video_gen_provider()``; the one selected by
-``video_gen.provider`` services every ``video_generate`` call. Providers live in
-``<repo>/plugins/video_gen/<name>/`` (built-in) or
-``~/.hermes/plugins/video_gen/<name>/`` (user, opt-in). Mirrors
-``agent/image_gen_provider.py`` so the two surfaces stay learnable together.
+Providers register via ``PluginContext.register_video_gen_provider()`` and live
+in ``<repo>/plugins/video_gen/<name>/`` (built-in) or
+``~/.hermes/plugins/video_gen/<name>/``; mirrors ``agent/image_gen_provider.py``.
+One tool covers text-to-video and image-to-video: ``image_url`` present routes to
+the provider's image-to-video endpoint. Video edit/extend are deliberately NOT
+exposed — backends are too inconsistent for one unified tool.
 
-One tool covers text-to-video and image-to-video: ``image_url`` present routes
-to the provider's image-to-video endpoint, absent routes to text-to-video. Users
-pick one model family; the provider picks the FAL/xAI endpoint. Video edit and
-extend are deliberately NOT exposed — backends are too inconsistent for one
-unified tool.
-
-Response shape (built by :func:`success_response` / :func:`error_response`)::
-
-    success         bool
-    video           str | None      URL or absolute file path
-    model           str             provider-specific model identifier
-    prompt          str             echoed prompt
-    modality        str             "text" | "image" (which mode was used)
-    aspect_ratio    str             provider-native (e.g. "16:9") or ""
-    duration        int             seconds (0 if not applicable)
-    provider        str             provider name (for diagnostics)
-    error           str             only when success=False
-    error_type      str             only when success=False
+Response shape (:func:`success_response` / :func:`error_response`): ``success``,
+``video`` (URL or absolute path), ``model``, ``prompt``, ``modality``
+("text" | "image"), ``aspect_ratio``, ``duration`` (seconds, 0 if n/a),
+``provider``; plus ``error`` / ``error_type`` only when ``success`` is False.
 """
 
 from __future__ import annotations
 
 import abc
 import logging
+import os
+import time
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -70,16 +56,10 @@ class VideoGenProvider(CatalogProviderBase):
         and the picker. Default fails closed: text-only, no optional features.
         """
         return {
-            "modalities": ["text"],
-            "aspect_ratios": list(COMMON_ASPECT_RATIOS),
-            "resolutions": list(COMMON_RESOLUTIONS),
-            "max_duration": 10,
-            "min_duration": 1,
-            "supports_audio": False,
-            "supports_negative_prompt": False,
-            "supports_seed": False,
-            "supports_upscale": False,
-            "max_reference_images": 0,
+            "modalities": ["text"], "aspect_ratios": list(COMMON_ASPECT_RATIOS),
+            "resolutions": list(COMMON_RESOLUTIONS), "max_duration": 10, "min_duration": 1,
+            "supports_audio": False, "supports_negative_prompt": False, "supports_seed": False,
+            "supports_upscale": False, "max_reference_images": 0,
         }
 
     @abc.abstractmethod
@@ -99,11 +79,6 @@ class VideoGenProvider(CatalogProviderBase):
         """
 
 
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-
 def save_b64_video(b64_data: str,*, prefix: str="video", extension: str="mp4") -> Path:
     """Decode base64 video data into ``$HERMES_HOME/cache/videos/``; return the path."""
     return provider_media.save_b64("videos", b64_data, prefix=prefix, extension=extension)
@@ -141,18 +116,11 @@ def success_response(
 ) -> Dict[str, Any]:
     """Uniform success dict; ``extra`` keys are added without overriding standard ones."""
     payload: Dict[str, Any] = {
-        "success": True,
-        "video": video,
-        "model": model,
-        "prompt": prompt,
-        "modality": modality,
-        "aspect_ratio": aspect_ratio,
-        "duration": int(duration) if duration else 0,
-        "provider": provider,
+        "success": True, "video": video, "model": model, "prompt": prompt, "modality": modality,
+        "aspect_ratio": aspect_ratio, "duration": int(duration) if duration else 0, "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
 
 
@@ -162,22 +130,11 @@ def error_response(
 ) -> Dict[str, Any]:
     """Build a uniform error response dict."""
     return {
-        "success": False,
-        "video": None,
-        "error": error,
-        "error_type": error_type,
-        "model": model,
-        "prompt": prompt,
-        "aspect_ratio": aspect_ratio,
-        "provider": provider,
+        "success": False, "video": None, "error": error, "error_type": error_type, "model": model,
+        "prompt": prompt, "aspect_ratio": aspect_ratio, "provider": provider,
     }
 
 
-# ---------------------------------------------------------------------------
-# Reusable OpenAI-compatible backend
-# ---------------------------------------------------------------------------
-
-
 class OpenAICompatibleVideoGenProvider(VideoGenProvider):
     """Generic text/image-to-video over the OpenAI ``client.videos`` API.
 
@@ -207,8 +164,6 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
     _poll_deadline_s: float = 900.0
 
     def _api_key(self) -> str:
-        import os
-
         return os.environ.get(self._env_key, "").strip()
 
     def is_available(self) -> bool:
@@ -217,8 +172,6 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
     def _create_and_poll(self, client: Any, call_kwargs: Dict[str, Any]) -> Any:
         """Create the job and poll to a terminal status (any); raise
         :class:`TimeoutError` when ``_poll_deadline_s`` passes first."""
-        import time
-
         video = client.videos.create(**call_kwargs)
         terminal = {"completed", "succeeded", "failed", "error", "cancelled", "canceled"}
         deadline = time.monotonic() + self._poll_deadline_s
@@ -234,10 +187,7 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
         return video
 
     def _base_url(self) -> str:
-        import os
-
-        override = os.environ.get(f"{self.name.upper()}_BASE_URL", "").strip()
-        return override or self._default_base_url
+        return os.environ.get(f"{self.name.upper()}_BASE_URL", "").strip() or self._default_base_url
 
     def generate(
         self, prompt: str, *, model: Optional[str] = None, image_url: Optional[str] = None,
@@ -247,13 +197,10 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
         seed: Optional[int] = None, **kwargs: Any,
     ) -> Dict[str, Any]:
         if not prompt or not prompt.strip():
-            return error_response(
-                error="prompt is required", error_type="invalid_request", provider=self.name
-            )
+            return error_response(error="prompt is required", error_type="invalid_request", provider=self.name)
         if not self._api_key():
             return error_response(
-                error=f"{self._env_key} is not set", error_type="missing_credentials",
-                provider=self.name,
+                error=f"{self._env_key} is not set", error_type="missing_credentials", provider=self.name
             )
         try:
             import openai
@@ -270,12 +217,17 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
                 error_type="no_model", provider=self.name,
             )
 
+        def fail(error: str, error_type: str) -> Dict[str, Any]:
+            return error_response(
+                error=error, error_type=error_type, provider=self.name, model=model_id, prompt=prompt,
+                aspect_ratio=aspect_ratio,
+            )
+
         # Fields ``videos.create`` doesn't name natively ride in ``extra_body``.
         extra_body = {
             k: v
             for k, v in {
-                "negative_prompt": negative_prompt,
-                "aspect_ratio": aspect_ratio,
+                "negative_prompt": negative_prompt, "aspect_ratio": aspect_ratio,
                 "image_url": image_url,  # presence ⇒ image-to-video
                 "seed": seed,
             }.items()
@@ -295,30 +247,22 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
                 video = self._create_and_poll(client, call_kwargs)
             except Exception as exc:  # noqa: BLE001 - surface any SDK/API/timeout failure uniformly
                 logger.debug("%s video generation failed", self.name, exc_info=True)
-                return error_response(
-                    error=f"{self.name} video generation failed: {exc}", error_type="api_error",
-                    provider=self.name, model=model_id, prompt=prompt, aspect_ratio=aspect_ratio,
-                )
+                return fail(f"{self.name} video generation failed: {exc}", "api_error")
 
             # DeepInfra reports "succeeded", OpenAI/Sora "completed" — accept both.
             status = getattr(video, "status", None)
             if status not in ("completed", "succeeded"):
                 # ``video.error`` is a pydantic object — str() keeps the dict JSON-serializable.
                 job_error = getattr(video, "error", None)
-                return error_response(
-                    error=str(job_error) if job_error else f"video job ended with status={status!r}",
-                    error_type="job_failed", provider=self.name, model=model_id, prompt=prompt,
-                    aspect_ratio=aspect_ratio,
-                )
+                return fail(str(job_error) if job_error else f"video job ended with status={status!r}", "job_failed")
 
             # Output is a delivery URL in ``data`` (DeepInfra/FAL) or only reachable
             # via the SDK download endpoint (OpenAI/Sora). Save locally either way —
             # DeepInfra's delivery URLs are short-lived.
             url = None
             for item in getattr(video, "data", None) or []:
-                candidate = item.get("url") if isinstance(item, dict) else getattr(item, "url", None)
-                if candidate:
-                    url = candidate
+                url = (item.get("url") if isinstance(item, dict) else getattr(item, "url", None)) or None
+                if url:
                     break
 
             try:
@@ -328,15 +272,10 @@ class OpenAICompatibleVideoGenProvider(VideoGenProvider):
                     raw = client.videos.download_content(video.id).read()
                     video_ref = str(save_bytes_video(raw, prefix=self.name))
             except Exception as exc:  # noqa: BLE001
-                if url:
-                    logger.debug("%s: saving video locally failed (%s); returning URL", self.name, exc)
-                    video_ref = url
-                else:
-                    return error_response(
-                        error=f"{self.name} video job succeeded but no output could be retrieved: {exc}",
-                        error_type="empty_response", provider=self.name, model=model_id,
-                        prompt=prompt, aspect_ratio=aspect_ratio,
-                    )
+                if not url:
+                    return fail(f"{self.name} video job succeeded but no output could be retrieved: {exc}", "empty_response")
+                logger.debug("%s: saving video locally failed (%s); returning URL", self.name, exc)
+                video_ref = url
 
             return success_response(
                 video=video_ref, model=model_id, prompt=prompt,

agent/video_gen_registry.py

@@ -1,25 +1,12 @@
-"""
-Video Generation Provider Registry
-==================================
+"""Video Generation Provider Registry.
 
-Central map of registered providers. Populated by plugins at import-time via
-``PluginContext.register_video_gen_provider()``; consumed by the
-``video_generate`` tool to dispatch each call to the active backend.
-
-Active selection
-----------------
-The active provider is chosen by ``video_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 return ``None`` (the tool surfaces a helpful error pointing
-   the user at ``hermes tools``).
-
-Mirrors ``agent/image_gen_registry.py``: the unconfigured fallback is
-filtered by ``is_available()`` so a box with credentials for only one backend
-(e.g. DeepInfra, while ``fal``/``xai`` register unconditionally) auto-selects
-it instead of returning ``None``. Unlike image gen there is no legacy ``fal``
-preference, and a configured-but-unregistered name fails closed.
+Populated by plugins via ``PluginContext.register_video_gen_provider()``;
+consumed by the ``video_generate`` tool. The active provider is
+``video_gen.provider`` from ``config.yaml``; a configured-but-unregistered name
+fails closed. If unset, the single *available* registered provider is used
+(mirrors ``agent/image_gen_registry.py`` minus its legacy ``fal`` preference)
+so a box with credentials for only one backend auto-selects it; otherwise None
+and the tool points the user at ``hermes tools``.
 """
 
 from __future__ import annotations
@@ -46,18 +33,14 @@ def get_active_provider() -> Optional[VideoGenProvider]:
 
     if configured:
         provider = snapshot.get(configured)
-        if provider is not None:
-            return provider
-        logger.debug(
-            "video_gen.provider='%s' configured but not registered; failing closed", configured
-        )
-        return None
+        if provider is None:
+            logger.debug(
+                "video_gen.provider='%s' configured but not registered; failing closed", configured
+            )
+        return provider
 
     available = [
         p for p in snapshot.values()
         if is_available_safe(p, logger, "video_gen provider %s.is_available() raised %s")
     ]
-    if len(available) == 1:
-        return available[0]
-
-    return None
+    return available[0] if len(available) == 1 else None

agent/vision_message_prep.py

@@ -2,7 +2,6 @@
 
 Vision capability probes, non-vision text fallbacks (cached ``vision_analyze`` descriptions), tool-result
 image stripping, and provider quirks (Anthropic dot preservation, Qwen portal message shaping).
-Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged.
 """
 import logging
 import asyncio
@@ -22,18 +21,30 @@ from utils import base_url_host_matches, base_url_hostname
 # Same logger name as the origin module so log records / caplog filters are unchanged.
 logger = logging.getLogger("run_agent")
 
+_IMAGE_PART_TYPES = {"image_url", "input_image"}
+_TEXT_PART_TYPES = {"text", "input_text"}
+_DATA_URL_SUFFIXES = {
+    "image/png": ".png", "image/gif": ".gif", "image/webp": ".webp", "image/jpeg": ".jpg", "image/jpg": ".jpg"
+}
+
+
+def _is_image_part(part: Any) -> bool:
+    return isinstance(part, dict) and part.get("type") in _IMAGE_PART_TYPES
+
 
 class VisionMessagePrepMixin:
     """Vision probes + image-part fallbacks for outgoing messages (see module docstring)."""
 
     @staticmethod
     def _content_has_image_parts(content: Any) -> bool:
-        if not isinstance(content, list):
-            return False
-        for part in content:
-            if isinstance(part, dict) and part.get("type") in {"image_url", "input_image"}:
-                return True
-        return False
+        return isinstance(content, list) and any(_is_image_part(part) for part in content)
+
+    def _provider_model_key(self) -> tuple[str, str]:
+        """``(provider.lower(), model)`` as recorded in ``_no_list_tool_content_models``."""
+        return (
+            (getattr(self, "provider", "") or "").strip().lower(),
+            (getattr(self, "model", "") or "").strip(),
+        )
 
     # 20 MB base64 ≈ 15 MB decoded — prevents OOM from an oversized data: URL in a shared gateway process.
     _MAX_DATA_URL_BASE64_BYTES = 20 * 1024 * 1024
@@ -51,13 +62,7 @@ class VisionMessagePrepMixin:
             mime_part = header[len("data:"):].split(";", 1)[0].strip()
             if mime_part.startswith("image/"):
                 mime = mime_part
-        suffix = {
-            "image/png": ".png",
-            "image/gif": ".gif",
-            "image/webp": ".webp",
-            "image/jpeg": ".jpg",
-            "image/jpg": ".jpg",
-        }.get(mime, ".jpg")
+        suffix = _DATA_URL_SUFFIXES.get(mime, ".jpg")
         tmp = tempfile.NamedTemporaryFile(prefix="anthropic_image_", suffix=suffix, delete=False)
         try:
             with tmp:
@@ -79,9 +84,7 @@ class VisionMessagePrepMixin:
         if cached:
             return cached
 
-        role_label = {
-            "assistant": "assistant", "tool": "tool result"
-        }.get(role, "user")
+        role_label = {"assistant": "assistant", "tool": "tool result"}.get(role, "user")
         analysis_prompt = (
             "Describe everything visible in this image in thorough detail. "
             "Include any text, code, UI, data, objects, people, layout, colors, "
@@ -111,14 +114,9 @@ class VisionMessagePrepMixin:
                 except OSError:
                     pass
 
-        if not description:
-            description = "Image analysis failed."
-
-        note = f"[The {role_label} attached an image. Here's what it contains:\n{description}]"
+        note = f"[The {role_label} attached an image. Here's what it contains:\n{description or 'Image analysis failed.'}]"
         if vision_source and not str(image_url or "").startswith("data:"):
-            note += (
-                f"\n[If you need a closer look, use vision_analyze with image_url: {vision_source}]"
-            )
+            note += f"\n[If you need a closer look, use vision_analyze with image_url: {vision_source}]"
 
         self._anthropic_image_fallback_cache[cache_key] = note
         return note
@@ -170,22 +168,15 @@ class VisionMessagePrepMixin:
             if not isinstance(part, dict):
                 continue
 
-            ptype = part.get("type")
-            if ptype in {"text", "input_text"}:
-                text = str(part.get("text", "") or "").strip()
-                if text:
-                    text_parts.append(text)
-                continue
-
-            if ptype in {"image_url", "input_image"}:
+            if _is_image_part(part):
                 image_data = part.get("image_url", {})
                 image_url = image_data.get("url", "") if isinstance(image_data, dict) else str(image_data or "")
-                if image_url:
-                    image_notes.append(self._describe_image_for_anthropic_fallback(image_url, role))
-                else:
-                    image_notes.append("[An image was attached but no image source was available.]")
+                image_notes.append(
+                    self._describe_image_for_anthropic_fallback(image_url, role) if image_url
+                    else "[An image was attached but no image source was available.]"
+                )
                 continue
-
+            # Text parts and unknown dict types both contribute their ``text``.
             text = str(part.get("text", "") or "").strip()
             if text:
                 text_parts.append(text)
@@ -194,11 +185,7 @@ class VisionMessagePrepMixin:
         suffix = "\n".join(text for text in text_parts if text).strip()
         if prefix and suffix:
             return f"{prefix}\n\n{suffix}"
-        if prefix:
-            return prefix
-        if suffix:
-            return suffix
-        return "[A multimodal message was converted to text for Anthropic compatibility.]"
+        return prefix or suffix or "[A multimodal message was converted to text for Anthropic compatibility.]"
 
     def _get_transport(self, api_mode: str = None):
         """Return the cached transport for the given (or current) api_mode (lazy; None if unregistered)."""
@@ -264,10 +251,7 @@ class VisionMessagePrepMixin:
                     tool_name, getattr(self, "provider", ""),
                 )
                 return _multimodal_text_summary(result)
-            key = (
-                (getattr(self, "provider", "") or "").strip().lower(),
-                (getattr(self, "model", "") or "").strip(),
-            )
+            key = self._provider_model_key()
             no_list = getattr(self, "_no_list_tool_content_models", None)
             if no_list and key in no_list:
                 logger.debug(
@@ -293,9 +277,7 @@ class VisionMessagePrepMixin:
         logger.warning(
             "Tool %s returned image content for non-vision model %s/%s; "
             "falling back to text summary",
-            tool_name,
-            self.provider,
-            self.model,
+            tool_name, self.provider, self.model,
         )
         return summary
 
@@ -315,10 +297,7 @@ class VisionMessagePrepMixin:
 
         if remember_model:
             # Record (provider, model) so we don't relearn this lesson.
-            key = (
-                (getattr(self, "provider", "") or "").strip().lower(),
-                (getattr(self, "model", "") or "").strip(),
-            )
+            key = self._provider_model_key()
             if not hasattr(self, "_no_list_tool_content_models"):
                 self._no_list_tool_content_models = set()
             if key[1]:  # only record when we actually have a model id
@@ -329,37 +308,26 @@ class VisionMessagePrepMixin:
             if not isinstance(msg, dict) or msg.get("role") != "tool":
                 continue
             content = msg.get("content")
-            if not isinstance(content, list):
+            # List content without image parts is left alone; stripping wouldn't reduce ambiguity.
+            if not self._content_has_image_parts(content):
                 continue
 
             # Salvage any text parts so the model still sees some signal.
             text_parts: List[str] = []
-            had_image = False
             for part in content:
-                if not isinstance(part, dict):
-                    if isinstance(part, str) and part.strip():
-                        text_parts.append(part.strip())
-                    continue
-                ptype = part.get("type")
-                if ptype == "image_url" or ptype == "input_image":
-                    had_image = True
-                    continue
-                if ptype in {"text", "input_text"}:
+                if isinstance(part, str):
+                    text = part.strip()
+                elif isinstance(part, dict) and part.get("type") in _TEXT_PART_TYPES:
                     text = str(part.get("text") or "").strip()
-                    if text:
-                        text_parts.append(text)
+                else:
+                    continue
+                if text:
+                    text_parts.append(text)
 
-            if not had_image:
-                # List content without image parts — leave alone; stripping wouldn't reduce ambiguity.
-                continue
-
-            if text_parts:
-                msg["content"] = "\n\n".join(text_parts)
-            else:
-                msg["content"] = (
-                    "[image content removed — provider does not accept "
-                    "list-type tool message content]"
-                )
+            msg["content"] = "\n\n".join(text_parts) or (
+                "[image content removed — provider does not accept "
+                "list-type tool message content]"
+            )
             changed = True
 
         return changed
@@ -371,10 +339,7 @@ class VisionMessagePrepMixin:
         dotted inference-profile IDs and rejects the hyphenated form with HTTP 400.
         """
         if (getattr(self, "provider", "") or "").lower() in {
-            "alibaba", "minimax", "minimax-cn",
-            "opencode-go", "opencode-zen",
-            "zai", "bedrock",
-            "xiaomi", "vertex",
+            "alibaba", "minimax", "minimax-cn", "opencode-go", "opencode-zen", "zai", "bedrock", "xiaomi", "vertex",
         }:
             return True
         base = (getattr(self, "base_url", "") or "").lower()
@@ -399,43 +364,14 @@ class VisionMessagePrepMixin:
         return base_url_host_matches(self._base_url_lower, "portal.qwen.ai")
 
     def _qwen_prepare_chat_messages(self, api_messages: list) -> list:
+        """Deep-copy ``api_messages`` and shape them for Qwen Portal (see the in-place variant)."""
         prepared = copy.deepcopy(api_messages)
-        if not prepared:
-            return prepared
-
-        for msg in prepared:
-            if not isinstance(msg, dict):
-                continue
-            content = msg.get("content")
-            if isinstance(content, str):
-                msg["content"] = [{"type": "text", "text": content}]
-            elif isinstance(content, list):
-                # Normalize: convert bare strings to text dicts, keep dicts as-is.
-                # deepcopy already created independent copies, no need for dict().
-                normalized_parts = []
-                for part in content:
-                    if isinstance(part, str):
-                        normalized_parts.append({"type": "text", "text": part})
-                    elif isinstance(part, dict):
-                        normalized_parts.append(part)
-                if normalized_parts:
-                    msg["content"] = normalized_parts
-
-        # Inject cache_control on the last part of the system message.
-        for msg in prepared:
-            if isinstance(msg, dict) and msg.get("role") == "system":
-                content = msg.get("content")
-                if isinstance(content, list) and content and isinstance(content[-1], dict):
-                    content[-1]["cache_control"] = {"type": "ephemeral"}
-                break
-
+        self._qwen_prepare_chat_messages_inplace(prepared)
         return prepared
 
     def _qwen_prepare_chat_messages_inplace(self, messages: list) -> None:
-        """In-place variant — mutates an already-copied message list."""
-        if not messages:
-            return
-
+        """Qwen Portal shaping, in place: every content becomes a list of parts (bare strings → text
+        dicts, dicts kept), then ``cache_control`` is injected on the last part of the system message."""
         for msg in messages:
             if not isinstance(msg, dict):
                 continue
@@ -443,12 +379,10 @@ class VisionMessagePrepMixin:
             if isinstance(content, str):
                 msg["content"] = [{"type": "text", "text": content}]
             elif isinstance(content, list):
-                normalized_parts = []
-                for part in content:
-                    if isinstance(part, str):
-                        normalized_parts.append({"type": "text", "text": part})
-                    elif isinstance(part, dict):
-                        normalized_parts.append(part)
+                normalized_parts = [
+                    {"type": "text", "text": part} if isinstance(part, str) else part
+                    for part in content if isinstance(part, (str, dict))
+                ]
                 if normalized_parts:
                     msg["content"] = normalized_parts
 

agent/web_search_provider.py

@@ -1,33 +1,22 @@
-"""
-Web Search Provider ABC
-=======================
+"""Web Search Provider ABC.
 
-Pluggable-backend interface for web search and content extraction — the SINGLE
-plugin-facing surface every in-tree web provider (brave-free, ddgs, searxng,
-exa, parallel, tavily, keenable, firecrawl) implements. Providers register via
-``PluginContext.register_web_search_provider()``; the active one (selected by
-``web.search_backend`` / ``web.extract_backend`` / ``web.backend``) services
-every ``web_search`` / ``web_extract`` call.
+The single plugin-facing surface every web provider (brave-free, ddgs, searxng,
+exa, parallel, tavily, keenable, firecrawl) implements; registered via
+``PluginContext.register_web_search_provider()`` and selected by
+``web.search_backend`` / ``web.extract_backend`` / ``web.backend``.
 
-Response shape (preserved from the legacy contract so the tool wrapper does not
-translate). Search::
+Response shapes (legacy contract, the tool wrapper does not translate)::
 
-    {"success": True, "data": {"web": [
-        {"title": str, "url": str, "description": str, "position": int}, ...]}}
-
-Extract::
-
-    {"success": True, "data": [
-        {"url": str, "title": str, "content": str, "raw_content": str, "metadata": dict}, ...]}
-
-On failure (either capability): ``{"success": False, "error": str}``.
+    search:  {"success": True, "data": {"web": [{"title", "url", "description", "position"}, ...]}}
+    extract: {"success": True, "data": [{"url", "title", "content", "raw_content", "metadata"}, ...]}
+    failure: {"success": False, "error": str}
 """
 
 from __future__ import annotations
 
 import abc
 import os
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List
 
 from agent.provider_base import ProviderBase
 
@@ -40,7 +29,6 @@ def get_provider_env(name: str) -> str:
     subprocess agent runs). Falls back to bare ``os.getenv`` when the config
     module is unavailable. Returns the stripped value, or ``""`` when unset.
     """
-    val: Optional[str] = None
     try:
         from hermes_cli.config import get_env_value
 

agent/web_search_registry.py

@@ -1,33 +1,19 @@
-"""
-Web Search Provider Registry
-============================
+"""Web Search Provider Registry.
 
-Central map of registered web providers. Populated by plugins at import-time
-via :meth:`PluginContext.register_web_search_provider`; consumed by the
-``web_search`` and ``web_extract`` tool wrappers in :mod:`tools.web_tools` to
-dispatch each call to the active backend.
+Populated by plugins via :meth:`PluginContext.register_web_search_provider`;
+consumed by the ``web_search`` / ``web_extract`` wrappers in :mod:`tools.web_tools`.
 
-Active selection
-----------------
-The active provider is chosen by configuration with this precedence:
+Active selection, in precedence order (the ``supports_search`` /
+``supports_extract`` capability filter applies at every step, so a search-only
+provider configured as ``web.extract_backend`` falls through):
 
-1. ``web.search_backend`` / ``web.extract_backend``
-   (per-capability override).
-2. ``web.backend`` (shared fallback).
-3. If exactly one capability-eligible provider is registered AND available,
-   use it.
-4. Legacy preference order — ``firecrawl`` → ``parallel`` → ``tavily`` →
-   ``exa`` → ``searxng`` → ``brave-free`` → ``ddgs`` — filtered by
-   availability. Matches the historic ``tools.web_tools._get_backend()``
-   candidate order so installs that never set a config key keep landing
-   on the same provider they did before the plugin migration.
-5. Otherwise ``None`` — the tool surfaces a helpful error pointing at
-   ``hermes tools``.
-
-The capability filter (``supports_search`` / ``supports_extract``) is
-applied at every step so a search-only provider (``brave-free``)
-configured as ``web.extract_backend`` correctly falls through to an
-extract-capable backend.
+1. ``web.search_backend`` / ``web.extract_backend``, then ``web.backend``.
+2. The single capability-eligible provider that is registered AND available.
+3. Legacy preference walk (``_LEGACY_PREFERENCE``) filtered by availability —
+   the historic ``tools.web_tools._get_backend()`` order, so installs that never
+   set a config key keep landing on the same provider.
+4. Keyless free-tier walk (``_KEYLESS_PREFERENCE``), last resort.
+5. ``None`` — the tool points the user at ``hermes tools``.
 """
 
 from __future__ import annotations
@@ -47,11 +33,6 @@ _registry: ProviderRegistry[WebSearchProvider] = ProviderRegistry(
 _registry.export(globals())
 
 
-# ---------------------------------------------------------------------------
-# Active-provider resolution
-# ---------------------------------------------------------------------------
-
-
 def _read_config_key(*path: str) -> Optional[str]:
     """Resolve a dotted config key from ``config.yaml``. Returns None on miss."""
     try:
@@ -75,38 +56,21 @@ def _configured_backend(capability: str) -> Optional[str]:
     return _read_config_key("web", f"{capability}_backend") or _read_config_key("web", "backend")
 
 
-# Legacy preference order — preserves behaviour for users who set no
-# ``web.backend`` / ``web.<capability>_backend`` config key at all. Matches
-# the historic candidate order in :func:`tools.web_tools._get_backend`
-# (paid providers first so existing paid setups don't get downgraded to
-# a free tier on upgrade). Filtered by ``is_available()`` at walk time so
-# we don't surface a provider the user has no credentials for.
-_LEGACY_PREFERENCE = (
-    "firecrawl", "parallel", "tavily", "exa", "searxng", "brave-free", "ddgs"
-)
+# Paid providers first so existing paid setups don't get downgraded to a free
+# tier on upgrade; filtered by ``is_available()`` at walk time.
+_LEGACY_PREFERENCE = ("firecrawl", "parallel", "tavily", "exa", "searxng", "brave-free", "ddgs")
 
-# Keyless free-tier walk — strictly LAST-resort, tried only after the
-# availability-filtered legacy walk finds nothing (i.e. the user has zero
-# web credentials and no importable ddgs). Ring vendors expose public
-# anonymous free tiers (see plugins/web/keyless_mcp.py). Unpinned keyless
-# traffic round-robins across the ring per request (the ring cursor lives
-# in keyless_mcp; an explicit `hermes tools` pick bypasses this walk
-# entirely, and rate-limited requests fail over to the next ring vendor).
-# Disable the tier with ``web.keyless_fallback: false``.
-_KEYLESS_PREFERENCE = (
-    "exa", "parallel", "firecrawl", "keenable"
-)
+# Anonymous public free tiers (see plugins/web/keyless_mcp.py); strictly last
+# resort, i.e. zero web credentials and no importable ddgs. Unpinned keyless
+# traffic round-robins across the ring per request; an explicit `hermes tools`
+# pick bypasses this walk. Disable with ``web.keyless_fallback: false``.
+_KEYLESS_PREFERENCE = ("exa", "parallel", "firecrawl", "keenable")
 
 
 def _keyless_preference() -> tuple:
-    """Return the keyless walk order for resolution.
-
-    Delegates the entry-vendor choice to the ring cursor in
-    :mod:`plugins.web.keyless_mcp` (round-robin per request, seeded by the
-    per-process random session id) so resolution and dispatch agree on
-    which vendor a fresh install starts at. The remaining vendors follow
-    in ring order as fallbacks for registration gaps.
-    """
+    """Keyless walk order, starting at the ring cursor in
+    :mod:`plugins.web.keyless_mcp` so resolution and dispatch agree on which
+    vendor a fresh install starts at; the rest follow in ring order."""
     try:
         from plugins.web.keyless_mcp import _KEYLESS_RING, _ring_cursor
 
@@ -133,9 +97,7 @@ def _resolve(configured: Optional[str], *, capability: str) -> Optional[WebSearc
     def _capable(p: WebSearchProvider) -> bool:
         if capability == "search":
             return bool(p.supports_search())
-        if capability == "extract":
-            return bool(p.supports_extract())
-        return False
+        return bool(p.supports_extract()) if capability == "extract" else False
 
     def _available(p: WebSearchProvider) -> bool:
         return is_available_safe(p, logger, "provider %s.is_available() raised %s")
@@ -198,16 +160,13 @@ def _keyless_tier_enabled() -> bool:
 
 def _disabled_web_plugin_for(configured: Optional[str] = None, *, capability: Optional[str] = None) -> Optional[str]:
     """Plugin key of a *disabled* bundled web plugin that would have provided
-    the configured backend (``web.<capability>_backend`` → ``web.backend``),
-    or None.
+    the configured backend (``web.<capability>_backend`` → ``web.backend``), or None.
 
-    Lets the dispatcher say "re-enable web-firecrawl" instead of a misleading
-    "No web extract provider configured" when the backend IS configured but
-    listed in ``plugins.disabled``. Resolving from config.yaml (rather than
-    the resolved backend) matters because a disabled provider fails the
-    availability gate and the dispatcher silently drops to the default.
-    Bundled web plugins live under ``web/<vendor>`` with the provider name
-    differing only by hyphen/underscore, so both sides are normalized.
+    Lets the dispatcher say "re-enable web-firecrawl" instead of "No web extract
+    provider configured". Resolved from config.yaml rather than the resolved
+    backend because a disabled provider fails the availability gate and silently
+    drops to the default. Bundled web plugins live under ``web/<vendor>`` with
+    the provider name differing only by hyphen/underscore, so both are normalized.
     """
     def _norm(s: str) -> str:
         return s.strip().lower().replace("-", "_")
@@ -223,14 +182,10 @@ def _disabled_web_plugin_for(configured: Optional[str] = None, *, capability: Op
 
         pm = get_plugin_manager()
         for key, loaded in pm._plugins.items():
-            if not isinstance(key, str) or not key.startswith("web/"):
-                continue
-            if loaded.enabled:
-                continue
-            if loaded.error != "disabled via config":
-                continue
-            vendor = key.split("/", 1)[1]
-            if _norm(vendor) == want:
+            if (
+                isinstance(key, str) and key.startswith("web/") and not loaded.enabled
+                and loaded.error == "disabled via config" and _norm(key.split("/", 1)[1]) == want
+            ):
                 return key
     except Exception as exc:  # noqa: BLE001 — diagnostics are best-effort
         logger.debug("disabled-web-plugin lookup failed: %s", exc)