refactor(tools): inline single-use helpers in write_approval, web_result_cache, xai_video_tools

tools/web_result_cache.py

@@ -14,6 +14,7 @@ import os
 import re
 import threading
 import time
+from contextlib import suppress
 from pathlib import Path
 from typing import Dict, Optional, Tuple
 from urllib.parse import urlparse
@@ -62,12 +63,6 @@ def normalize_query(query: str) -> str:
     return re.sub(r"\s+", " ", (query or "").strip().lower())
 
 
-def _host_slug(url: str) -> str:
-    """Filesystem-safe hostname slug for cache filenames (``"page"`` when hostless)."""
-    host = (urlparse(url).hostname or "page").replace(":", "_")
-    return re.sub(r"[^A-Za-z0-9._-]", "-", host)[:60].strip("-") or "page"
-
-
 def _deep_copy(response: dict) -> dict:
     """Defensive copy so callers mutating a hit never corrupt the cached entry."""
     return json.loads(json.dumps(response))
@@ -200,10 +195,10 @@ def _entry_file_path(url: str, format: Optional[str], provider: str) -> Optional
     (keyed on URL alone), which html/markdown or two providers' copies of one URL would overwrite."""
     if (d := _cache_dir()) is None:
         return None
-    try:
-        slug = _host_slug(url)
-    except Exception:  # noqa: BLE001
-        slug = "page"
+    slug = "page"  # filesystem-safe hostname slug; "page" when hostless/unparseable
+    with suppress(Exception):
+        host = (urlparse(url).hostname or "page").replace(":", "_")
+        slug = re.sub(r"[^A-Za-z0-9._-]", "-", host)[:60].strip("-") or "page"
     return d / f"{slug}-{_url_digest(url, format, provider)}.cache.md"
 
 

tools/write_approval.py

@@ -65,10 +65,6 @@ def _pending_path(subsystem: str, pending_id: str) -> Path:
     return get_hermes_home() / "pending" / subsystem / f"{pending_id}.json"
 
 
-def _read_record(path: Path) -> Dict[str, Any]:
-    return json.loads(path.read_text(encoding="utf-8"))
-
-
 def _pending_files(subsystem: str) -> list:
     d = _pending_path(subsystem, "").parent
     return list(d.glob("*.json")) if d.exists() else []
@@ -101,7 +97,7 @@ def list_pending(subsystem: str) -> List[Dict[str, Any]]:
     records: List[Dict[str, Any]] = []
     for p in _pending_files(subsystem):
         try:
-            records.append(_read_record(p))
+            records.append(json.loads(p.read_text(encoding="utf-8")))
         except Exception:
             logger.warning("Skipping unreadable pending record: %s", p)
     records.sort(key=lambda r: r.get("created_at", 0))
@@ -112,7 +108,7 @@ def get_pending(subsystem: str, pending_id: str) -> Optional[Dict[str, Any]]:
     """Return a single pending record by id, or None."""
     with suppress(Exception):
         path = _pending_path(subsystem, pending_id)
-        return _read_record(path) if path.exists() else None
+        return json.loads(path.read_text(encoding="utf-8")) if path.exists() else None
     return None
 
 

tools/xai_video_tools.py

@@ -37,12 +37,6 @@ def _coerce_int(value: Any) -> Optional[int]:
         return None
 
 
-def _normalize_public_video_url(video_url: Any) -> Optional[str]:
-    """Require a public HTTPS MP4 URL (``http``/``https`` only)."""
-    cleaned = _clean_string(video_url)
-    return cleaned if cleaned and cleaned.lower().startswith(("http://", "https://")) else None
-
-
 _VIDEO_URL_PARAM = {
     "type": "string",
     "description": (
@@ -94,11 +88,10 @@ XAI_VIDEO_EXTEND_SCHEMA: Dict[str, Any] = _xai_video_schema(
 
 
 def _run_xai_video_tool(args: Dict[str, Any], op: str, run, **extra: Any) -> str:
-    prompt = _clean_string(args.get("prompt"))
-    video_url = _normalize_public_video_url(args.get("video_url"))
+    prompt, video_url = _clean_string(args.get("prompt")), _clean_string(args.get("video_url"))
     if not prompt:
         return tool_error(f"prompt is required for xAI video {op}")
-    if not video_url:
+    if not (video_url and video_url.lower().startswith(("http://", "https://"))):  # public URL only
         return tool_error(
             "video_url must be a public HTTPS MP4 URL (the `video`/`public_url` "
             "from a prior Imagine result)"
@@ -122,8 +115,7 @@ def _handle_xai_video_edit(args: Dict[str, Any], **_kw: Any) -> str:
 
 
 def _handle_xai_video_extend(args: Dict[str, Any], **_kw: Any) -> str:
-    duration = _coerce_int(args.get("duration"))
-    return _run_xai_video_tool(args, "extend", run_xai_video_extend, duration=duration)
+    return _run_xai_video_tool(args, "extend", run_xai_video_extend, duration=_coerce_int(args.get("duration")))
 
 
 for _name, _schema, _handler in (

tools/yuanbao_tools.py

@@ -62,8 +62,7 @@ def _get_active_adapter():
 
 
 def _adapter():
-    adapter = _get_active_adapter()
-    if adapter is None:
+    if (adapter := _get_active_adapter()) is None:
         raise _YbError("Yuanbao adapter is not connected")
     return adapter