From 8b53adec12428f63aee37aa7ef132d3b1ed194c4 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:05:42 -0700
Subject: [PATCH 1/5] refactor(tools): yuanbao handlers take args directly;
 shared _yb_tool error envelope decorator

---
 tools/yuanbao_tools.py | 462 +++++++++++++++++------------------------
 1 file changed, 189 insertions(+), 273 deletions(-)

diff --git a/tools/yuanbao_tools.py b/tools/yuanbao_tools.py
index 155fa7d430..b7b26407b1 100644
--- a/tools/yuanbao_tools.py
+++ b/tools/yuanbao_tools.py
@@ -8,9 +8,10 @@ The active adapter singleton lives in ``gateway.platforms.yuanbao.get_active_ada
 
 from __future__ import annotations
 
+import functools
 import logging
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import Tuple
 
 from tools.registry import registry, tool_result
 
@@ -26,6 +27,36 @@ MENTION_HINT = (
 # Image extensions for media dispatch (mirrors MessageSender.IMAGE_EXTS)
 _IMAGE_EXTS = frozenset({".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"})
 
+
+class _YbError(Exception):
+    """Tool-level failure; ``payload`` is the error envelope returned to the model."""
+
+    def __init__(self, msg: str, **extra):
+        super().__init__(msg)
+        self.payload = {"success": False, "error": msg, **extra}
+
+
+def _err(msg: str) -> dict:
+    return {"success": False, "error": msg}
+
+
+def _yb_tool(label: str):
+    """Handler decorator: ``fn(args)`` result → tool_result; ``_YbError`` → its envelope;
+    any other exception → logged + ``_err(str(exc))``."""
+    def deco(fn):
+        @functools.wraps(fn)
+        async def handler(args, **kw):
+            try:
+                return tool_result(await fn(args))
+            except _YbError as exc:
+                return tool_result(exc.payload)
+            except Exception as exc:
+                logger.exception("[yuanbao_tools] %s error", label)
+                return tool_result(_err(str(exc)))
+        return handler
+    return deco
+
+
 def _get_active_adapter():
     """Lazy import to avoid ImportError when gateway.platforms.yuanbao is unavailable."""
     try:
@@ -35,333 +66,194 @@ def _get_active_adapter():
         return None
 
 
-def _err(msg: str) -> dict:
-    return {"success": False, "error": msg}
+def _adapter():
+    adapter = _get_active_adapter()
+    if adapter is None:
+        raise _YbError("Yuanbao adapter is not connected")
+    return adapter
+
+
+def _session_env(name: str) -> str:
+    try:
+        from gateway.session_context import get_session_env
+        return get_session_env(name, "")
+    except Exception:
+        return ""
 
 
 def _nick(m: dict) -> str:
     return m.get("nickname", m.get("nick_name", ""))
 
 
-async def _resolve_dm_recipient(adapter, group_code: str, name: str) -> Tuple[str, str, Optional[dict]]:
+async def _members(adapter, group_code: str) -> list:
+    raw = await adapter.get_group_member_list(group_code)
+    if raw is None:
+        raise _YbError("get_group_member_list returned None")
+    return raw.get("members", [])
+
+
+async def _resolve_dm_recipient(adapter, group_code: str, name: str) -> Tuple[str, str]:
     """Resolve ``name`` to (user_id, nickname) via the group member list.
 
-    Returns ``("", "", error_dict)`` on failure; >1 partial match yields candidates
-    for disambiguation instead of guessing.
+    >1 partial match raises with ``candidates`` for disambiguation instead of guessing.
     """
     if not group_code:
-        return "", "", _err("group_code is required when user_id is not provided")
+        raise _YbError("group_code is required when user_id is not provided")
     if not name:
-        return "", "", _err("name is required when user_id is not provided")
-    try:
-        raw = await adapter.get_group_member_list(group_code)
-        if raw is None:
-            return "", "", _err("get_group_member_list returned None")
-        filt = name.strip().lower()
-        matched = [
-            m for m in raw.get("members", [])
-            if filt in (m.get("nickname") or m.get("nick_name") or "").lower()
-        ]
-        if not matched:
-            return "", "", _err(f'No member matching "{name}" found in group {group_code}.')
-        if len(matched) > 1:
-            return "", "", {
-                "success": False,
-                "error": f'Multiple members match "{name}". Please specify which one.',
-                "candidates": [
-                    {"user_id": m.get("user_id", ""), "nickname": _nick(m)} for m in matched
-                ],
-            }
-        m = matched[0]
-        return m.get("user_id", ""), m.get("nickname", m.get("nick_name", name)), None
-    except Exception as exc:
-        logger.exception("[yuanbao_tools] send_dm member lookup error")
-        return "", "", _err(str(exc))
+        raise _YbError("name is required when user_id is not provided")
+    filt = name.strip().lower()
+    matched = [
+        m for m in await _members(adapter, group_code)
+        if filt in (m.get("nickname") or m.get("nick_name") or "").lower()
+    ]
+    if not matched:
+        raise _YbError(f'No member matching "{name}" found in group {group_code}.')
+    if len(matched) > 1:
+        raise _YbError(
+            f'Multiple members match "{name}". Please specify which one.',
+            candidates=[{"user_id": m.get("user_id", ""), "nickname": _nick(m)} for m in matched],
+        )
+    m = matched[0]
+    return m.get("user_id", ""), m.get("nickname", m.get("nick_name", name))
 
 
-async def get_group_info(group_code: str) -> dict:
+@_yb_tool("get_group_info")
+async def get_group_info(args) -> dict:
     """查询群基本信息（群名、群主、成员数）。"""
+    group_code = args.get("group_code", "")
     if not group_code:
         return _err("group_code is required")
-    adapter = _get_active_adapter()
-    if adapter is None:
-        return _err("Yuanbao adapter is not connected")
-    try:
-        gi = await adapter.query_group_info(group_code)
-        if gi is None:
-            return _err("query_group_info returned None")
-        return {
-            "success": True,
-            "group_code": group_code,
-            "group_name": gi.get("group_name", ""),
-            "member_count": gi.get("member_count", 0),
-            "owner": {
-                "user_id": gi.get("owner_id", ""),
-                "nickname": gi.get("owner_nickname", ""),
-            },
-            "note": 'The group is called "派 (Pai)" in the app.',
-        }
-    except Exception as exc:
-        logger.exception("[yuanbao_tools] get_group_info error")
-        return _err(str(exc))
+    gi = await _adapter().query_group_info(group_code)
+    if gi is None:
+        return _err("query_group_info returned None")
+    return {
+        "success": True,
+        "group_code": group_code,
+        "group_name": gi.get("group_name", ""),
+        "member_count": gi.get("member_count", 0),
+        "owner": {"user_id": gi.get("owner_id", ""), "nickname": gi.get("owner_nickname", "")},
+        "note": 'The group is called "派 (Pai)" in the app.',
+    }
 
 
-async def query_group_members(
-    group_code: str,
-    action: str = "list_all",
-    name: str = "",
-    mention: bool = False,
-) -> dict:
+@_yb_tool("query_group_members")
+async def query_group_members(args) -> dict:
     """统一的群成员查询（对齐 TS query_session_members）。
 
     action: find (按昵称模糊搜索; 无 name 时等同 list_all) / list_bots / list_all (默认).
     """
+    group_code, action, name = args.get("group_code", ""), args.get("action", "list_all"), args.get("name", "")
     if not group_code:
         return _err("group_code is required")
-    adapter = _get_active_adapter()
-    if adapter is None:
-        return _err("Yuanbao adapter is not connected")
-    try:
-        raw = await adapter.get_group_member_list(group_code)
-        if raw is None:
-            return _err("get_group_member_list returned None")
+    all_members = [
+        {
+            "user_id": m.get("user_id", ""),
+            "nickname": _nick(m),
+            "role": _USER_TYPE_LABEL.get(m.get("user_type", m.get("role", 0)), "unknown"),
+        }
+        for m in await _members(_adapter(), group_code)
+    ]
+    if not all_members:
+        return _err("No members found in this group.")
 
-        all_members = [
-            {
-                "user_id": m.get("user_id", ""),
-                "nickname": _nick(m),
-                "role": _USER_TYPE_LABEL.get(m.get("user_type", m.get("role", 0)), "unknown"),
-            }
-            for m in raw.get("members", [])
-        ]
-        if not all_members:
-            return _err("No members found in this group.")
+    hint = {"mention_hint": MENTION_HINT} if args.get("mention", False) else {}
 
-        hint = {"mention_hint": MENTION_HINT} if mention else {}
+    def _listing(ok: bool, msg: str, members: list) -> dict:
+        return {"success": ok, "msg": msg, "members": members, **hint}
 
-        def _listing(ok: bool, msg: str, members: list) -> dict:
-            return {"success": ok, "msg": msg, "members": members, **hint}
+    if action == "list_bots":
+        bots = [m for m in all_members if m["role"] in {"yuanbao_ai", "bot"}]
+        if not bots:
+            return _err("No bots found in this group.")
+        return _listing(True, f"Found {len(bots)} bot(s).", bots)
 
-        if action == "list_bots":
-            bots = [m for m in all_members if m["role"] in {"yuanbao_ai", "bot"}]
-            if not bots:
-                return _err("No bots found in this group.")
-            return _listing(True, f"Found {len(bots)} bot(s).", bots)
+    if action == "find" and name:
+        filt = name.strip().lower()
+        matched = [m for m in all_members if filt in m["nickname"].lower()]
+        if matched:
+            return _listing(True, f'Found {len(matched)} member(s) matching "{name}".', matched)
+        return _listing(False, f'No match for "{name}". All members listed below.', all_members)
 
-        if action == "find" and name:
-            filt = name.strip().lower()
-            matched = [m for m in all_members if filt in m["nickname"].lower()]
-            if matched:
-                return _listing(True, f'Found {len(matched)} member(s) matching "{name}".', matched)
-            return _listing(False, f'No match for "{name}". All members listed below.', all_members)
-
-        return _listing(True, f"Found {len(all_members)} member(s).", all_members)
-    except Exception as exc:
-        logger.exception("[yuanbao_tools] query_group_members error")
-        return _err(str(exc))
+    return _listing(True, f"Found {len(all_members)} member(s).", all_members)
 
 
-async def search_sticker(query: str = "", limit: int = 10) -> dict:
+@_yb_tool("search_sticker")
+async def search_sticker(args) -> dict:
     """在内置贴纸表中按关键词模糊搜索，返回 Top-N 候选（空 query 返回前 N 条）。"""
     from gateway.platforms.yuanbao_sticker import search_stickers
 
+    query, limit = args.get("query", ""), args.get("limit", 10)
     try:
         safe_limit = max(1, min(50, int(limit) if limit else 10))
     except (TypeError, ValueError):
         safe_limit = 10
-
-    try:
-        matches = search_stickers(query or "", limit=safe_limit)
-    except Exception as exc:
-        logger.exception("[yuanbao_tools] search_sticker error")
-        return _err(str(exc))
-
+    matches = search_stickers(query or "", limit=safe_limit)
     return {
         "success": True,
         "query": query or "",
         "count": len(matches),
         "results": [
-            {
-                "sticker_id": s.get("sticker_id", ""),
-                "name": s.get("name", ""),
-                "description": s.get("description", ""),
-                "package_id": s.get("package_id", ""),
-            }
+            {k: s.get(k, "") for k in ("sticker_id", "name", "description", "package_id")}
             for s in matches
         ],
     }
 
 
-async def send_sticker(
-    sticker: str = "",
-    chat_id: str = "",
-    reply_to: str = "",
-) -> dict:
+@_yb_tool("send_sticker")
+async def send_sticker(args) -> dict:
     """向 chat_id（缺省取当前会话 HERMES_SESSION_CHAT_ID）发送一张内置贴纸（TIMFaceElem）。
 
     ``sticker``: 名称（如 "六六六"）或 sticker_id（如 "278"）；为空时随机发送。
     ``chat_id``: ``direct:{account_id}`` / ``group:{group_code}`` / 裸 account_id。
     """
-    from gateway.session_context import get_session_env
-    from gateway.platforms.yuanbao_sticker import (
-        get_sticker_by_id,
-        get_sticker_by_name,
-        get_random_sticker,
-    )
+    from gateway.platforms.yuanbao_sticker import get_sticker_by_id, get_sticker_by_name, get_random_sticker
 
-    target = (chat_id or "").strip() or get_session_env("HERMES_SESSION_CHAT_ID", "")
+    target = (args.get("chat_id", "") or "").strip() or _session_env("HERMES_SESSION_CHAT_ID")
     if not target:
         return _err("chat_id is required (no active yuanbao session detected)")
-    adapter = _get_active_adapter()
-    if adapter is None:
-        return _err("Yuanbao adapter is not connected")
+    adapter = _adapter()
 
-    raw = (sticker or "").strip()
-    sticker_obj: Optional[dict] = None
+    raw = (args.get("sticker", "") or "").strip()
     if not raw:
         sticker_obj = get_random_sticker()
     else:
-        if raw.isdigit():
-            sticker_obj = get_sticker_by_id(raw)
+        sticker_obj = get_sticker_by_id(raw) if raw.isdigit() else None
         if sticker_obj is None:
             sticker_obj = get_sticker_by_name(raw)
     if sticker_obj is None:
-        return _err(
-            f"Sticker not found: {raw!r}. "
-            f"Use search_sticker first to discover available stickers."
-        )
-
-    try:
-        result = await adapter.send_sticker(
-            chat_id=target,
-            sticker_name=sticker_obj.get("name", ""),
-            reply_to=reply_to or None,
-        )
-    except Exception as exc:
-        logger.exception("[yuanbao_tools] send_sticker error")
-        return _err(str(exc))
+        return _err(f"Sticker not found: {raw!r}. Use search_sticker first to discover available stickers.")
 
+    result = await adapter.send_sticker(
+        chat_id=target, sticker_name=sticker_obj.get("name", ""), reply_to=args.get("reply_to", "") or None,
+    )
     if getattr(result, "success", False):
         return {
             "success": True,
             "chat_id": target,
-            "sticker": {
-                "sticker_id": sticker_obj.get("sticker_id", ""),
-                "name": sticker_obj.get("name", ""),
-            },
+            "sticker": {"sticker_id": sticker_obj.get("sticker_id", ""), "name": sticker_obj.get("name", "")},
             "message_id": getattr(result, "message_id", None),
             "note": "Sticker delivered to the chat. If you have additional text to say, reply now; otherwise end your turn without generating text.",
         }
     return _err(getattr(result, "error", "send_sticker failed"))
 
 
-async def send_dm(
-    group_code: str,
-    name: str,
-    message: str,
-    user_id: str = "",
-    media_files: Optional[List[Tuple[str, bool]]] = None,
-) -> dict:
+@_yb_tool("send_dm")
+async def send_dm(args) -> dict:
     """Send a DM to a group member, with optional media.
 
-    Without ``user_id`` the member list of ``group_code`` is searched by ``name``
-    (partial, case-insensitive; >1 match returns candidates). Text goes via
-    adapter.send_dm; media_files (path, is_voice) go via send_image_file (images) or
-    send_document. Partial media failures are reported in ``note``, not as failure.
+    group_code: explicit arg, else the session's "group:<code>" chat_id. Without ``user_id``
+    the member list is searched by ``name`` (partial, case-insensitive; >1 match returns
+    candidates). media_files items are {"path", "is_voice"} dicts or (path, is_voice) pairs;
+    ``MEDIA:<path>`` tags embedded in the text are extracted too. Text goes via adapter.send_dm;
+    media via send_image_file (images) or send_document. Partial media failures are reported
+    in ``note``, not as failure.
     """
-    if not message and not media_files:
-        return _err("message or media_files is required")
-    adapter = _get_active_adapter()
-    if adapter is None:
-        return _err("Yuanbao adapter is not connected")
-
-    resolved_user_id = user_id.strip() if user_id else ""
-    resolved_nickname = name.strip()
-
-    if not resolved_user_id:
-        resolved_user_id, resolved_nickname, err = await _resolve_dm_recipient(adapter, group_code, name)
-        if err is not None:
-            return err
-
-    if not resolved_user_id:
-        return _err("Could not resolve user_id")
-
-    chat_id = f"direct:{resolved_user_id}"
-    last_result = None
-    errors: list[str] = []
-    try:
-        if message and message.strip():
-            last_result = await adapter.send_dm(resolved_user_id, message, group_code=group_code)
-            if not last_result.success:
-                errors.append(last_result.error or "text send failed")
-
-        for media_path, _is_voice in media_files or []:
-            send = adapter.send_image_file if Path(media_path).suffix.lower() in _IMAGE_EXTS else adapter.send_document
-            last_result = await send(chat_id, media_path, group_code=group_code)
-            if not last_result.success:
-                errors.append(last_result.error or "media send failed")
-
-        if last_result is None:
-            return _err("No deliverable text or media remained")
-        if errors and not last_result.success:
-            return _err("; ".join(errors))
-
-        result = {
-            "success": True,
-            "user_id": resolved_user_id,
-            "nickname": resolved_nickname,
-            "message_id": last_result.message_id,
-            "note": f'DM sent to "{resolved_nickname}" successfully.',
-        }
-        if errors:
-            result["note"] += f" (partial failure: {'; '.join(errors)})"
-        return result
-    except Exception as exc:
-        logger.exception("[yuanbao_tools] send_dm error")
-        return _err(str(exc))
-
-
-# ---------------------------------------------------------------------------
-# Registry registration
-# ---------------------------------------------------------------------------
-
-def _check_yuanbao():
-    """Toolset availability check — True when running in a yuanbao gateway session."""
-    try:
-        from gateway.session_context import get_session_env
-        if get_session_env("HERMES_SESSION_PLATFORM", "") == "yuanbao":
-            return True
-    except Exception:
-        pass
-    return _get_active_adapter() is not None
-
-
-async def _handle_yb_query_group_info(args, **kw):
-    return tool_result(await get_group_info(group_code=args.get("group_code", "")))
-
-
-async def _handle_yb_query_group_members(args, **kw):
-    return tool_result(await query_group_members(
-        group_code=args.get("group_code", ""),
-        action=args.get("action", "list_all"),
-        name=args.get("name", ""),
-        mention=bool(args.get("mention", False)),
-    ))
-
-
-async def _handle_yb_send_dm(args, **kw):
-    # group_code: explicit arg, else the session's "group:<code>" chat_id.
     group_code = args.get("group_code", "")
     if not group_code:
-        try:
-            from gateway.session_context import get_session_env
-            chat_id = get_session_env("HERMES_SESSION_CHAT_ID", "")
-            if chat_id.startswith("group:"):
-                group_code = chat_id.split(":", 1)[1]
-        except Exception:
-            pass
+        chat_id = _session_env("HERMES_SESSION_CHAT_ID")
+        if chat_id.startswith("group:"):
+            group_code = chat_id.split(":", 1)[1]
 
-    # media_files items: {"path", "is_voice"} dicts or (path, is_voice) pairs.
     media_files = []
     for item in args.get("media_files") or []:
         if isinstance(item, dict):
@@ -369,32 +261,56 @@ async def _handle_yb_send_dm(args, **kw):
         elif isinstance(item, (list, tuple)) and len(item) >= 2:
             media_files.append((str(item[0]), bool(item[1])))
 
-    # LLMs often embed MEDIA:<path> tags in the text instead of using media_files.
     from gateway.platforms.base import BasePlatformAdapter
     embedded_media, message = BasePlatformAdapter.extract_media(args.get("message", ""))
-    if embedded_media:
-        media_files.extend(embedded_media)
+    media_files.extend(embedded_media or [])
     media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
 
-    return tool_result(await send_dm(
-        group_code=group_code,
-        name=args.get("name", ""),
-        message=message,
-        user_id=args.get("user_id", ""),
-        media_files=media_files or None,
-    ))
+    if not message and not media_files:
+        return _err("message or media_files is required")
+    adapter = _adapter()
+
+    name, user_id = args.get("name", ""), args.get("user_id", "")
+    resolved_user_id, resolved_nickname = user_id.strip() if user_id else "", name.strip()
+    if not resolved_user_id:
+        resolved_user_id, resolved_nickname = await _resolve_dm_recipient(adapter, group_code, name)
+    if not resolved_user_id:
+        return _err("Could not resolve user_id")
+
+    chat_id = f"direct:{resolved_user_id}"
+    last_result = None
+    errors: list[str] = []
+    if message and message.strip():
+        last_result = await adapter.send_dm(resolved_user_id, message, group_code=group_code)
+        if not last_result.success:
+            errors.append(last_result.error or "text send failed")
+
+    for media_path, _is_voice in media_files:
+        send = adapter.send_image_file if Path(media_path).suffix.lower() in _IMAGE_EXTS else adapter.send_document
+        last_result = await send(chat_id, media_path, group_code=group_code)
+        if not last_result.success:
+            errors.append(last_result.error or "media send failed")
+
+    if last_result is None:
+        return _err("No deliverable text or media remained")
+    if errors and not last_result.success:
+        return _err("; ".join(errors))
+
+    result = {
+        "success": True,
+        "user_id": resolved_user_id,
+        "nickname": resolved_nickname,
+        "message_id": last_result.message_id,
+        "note": f'DM sent to "{resolved_nickname}" successfully.',
+    }
+    if errors:
+        result["note"] += f" (partial failure: {'; '.join(errors)})"
+    return result
 
 
-async def _handle_yb_search_sticker(args, **kw):
-    return tool_result(await search_sticker(query=args.get("query", ""), limit=args.get("limit", 10)))
-
-
-async def _handle_yb_send_sticker(args, **kw):
-    return tool_result(await send_sticker(
-        sticker=args.get("sticker", ""),
-        chat_id=args.get("chat_id", ""),
-        reply_to=args.get("reply_to", ""),
-    ))
+def _check_yuanbao():
+    """Toolset availability check — True when running in a yuanbao gateway session."""
+    return _session_env("HERMES_SESSION_PLATFORM") == "yuanbao" or _get_active_adapter() is not None
 
 
 # (schema, handler, emoji); the tool name is schema["name"].
@@ -417,7 +333,7 @@ _TOOLS = (
                 "required": ["group_code"],
             },
         },
-        _handle_yb_query_group_info,
+        get_group_info,
         "👥",
     ),
     (
@@ -464,7 +380,7 @@ _TOOLS = (
                 "required": ["group_code", "action"],
             },
         },
-        _handle_yb_query_group_members,
+        query_group_members,
         "📋",
     ),
     (
@@ -532,7 +448,7 @@ _TOOLS = (
                 "required": [],
             },
         },
-        _handle_yb_send_dm,
+        send_dm,
         "✉️",
     ),
     (
@@ -563,7 +479,7 @@ _TOOLS = (
                 "required": [],
             },
         },
-        _handle_yb_search_sticker,
+        search_sticker,
         "🔍",
     ),
     (
@@ -606,7 +522,7 @@ _TOOLS = (
                 "required": [],
             },
         },
-        _handle_yb_send_sticker,
+        send_sticker,
         "🎨",
     ),
 )

From dc481dcb37355286d9fc44a77c9f6d8186c70d45 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:15:04 -0700
Subject: [PATCH 2/5] refactor(tools): x_search filter/tool_def builder helper;
 single retry loop; inline one-shot config getters

---
 tools/x_search_tool.py | 247 ++++++++++++++---------------------------
 tools/yuanbao_tools.py |   3 +-
 2 files changed, 88 insertions(+), 162 deletions(-)

diff --git a/tools/x_search_tool.py b/tools/x_search_tool.py
index 9c167a34cb..fabce55fe1 100644
--- a/tools/x_search_tool.py
+++ b/tools/x_search_tool.py
@@ -4,16 +4,13 @@
 Registers when either xAI credential path is available (``XAI_API_KEY`` or
 ``hermes auth add xai-oauth``). At call time an explicit ``XAI_API_KEY`` wins
 (``prefer_api_key=True``): x_search is API-metered and the subscription OAuth
-bearer answers ``/v1/responses`` in a degraded no-citation mode (#88040).
+bearer answers ``/v1/responses`` in a degraded no-citation mode.
 
-Defensive output: ``from_date``/``to_date`` are validated client-side (strict
-``YYYY-MM-DD``, ``from <= to``, ``from`` not in the future) so malformed windows
-fail fast instead of burning a billable call. Successful responses carry
-``degraded``/``degraded_reason``: True when a narrowing filter was active AND
-xAI returned no citations in either channel, meaning the answer came from the
-model's own knowledge rather than the X index.
-
-Salvaged from PR #10786 (originally by @Jaaneek).
+``from_date``/``to_date`` are validated client-side (strict ``YYYY-MM-DD``, ``from <= to``,
+``from`` not in the future) so malformed windows fail fast instead of burning a billable
+call. Successful responses carry ``degraded``/``degraded_reason``: True when a narrowing
+filter was active AND xAI returned no citations in either channel (answer came from the
+model's own knowledge, not the X index).
 """
 
 from __future__ import annotations
@@ -38,10 +35,6 @@ X_SEARCH_REASONING_EFFORTS = ("low", "medium", "high", "xhigh")
 MAX_HANDLES = 10
 
 
-# ---------------------------------------------------------------------------
-# Config
-# ---------------------------------------------------------------------------
-
 def _load_x_search_config() -> Dict[str, Any]:
     try:
         from hermes_cli.config import load_config
@@ -51,29 +44,20 @@ def _load_x_search_config() -> Dict[str, Any]:
         return {}
 
 
-def _get_x_search_model() -> str:
-    return str(_load_x_search_config().get("model") or "").strip() or DEFAULT_X_SEARCH_MODEL
-
-
 def _get_x_search_reasoning_effort() -> Optional[str]:
     raw_value = _load_x_search_config().get("reasoning_effort")
     if raw_value is None or not str(raw_value).strip():
         return None
-
     effort = str(raw_value).strip().lower()
     if effort not in X_SEARCH_REASONING_EFFORTS:
         allowed = ", ".join(X_SEARCH_REASONING_EFFORTS)
-        raise ValueError(
-            f"x_search.reasoning_effort must be one of: {allowed} "
-            f"(got {raw_value!r})"
-        )
+        raise ValueError(f"x_search.reasoning_effort must be one of: {allowed} (got {raw_value!r})")
     return effort
 
 
 def _get_x_search_int(key: str, default: int, floor: int) -> int:
-    raw_value = _load_x_search_config().get(key, default)
     try:
-        return max(floor, int(raw_value))
+        return max(floor, int(_load_x_search_config().get(key, default)))
     except Exception:
         return default
 
@@ -82,20 +66,12 @@ def _get_x_search_timeout_seconds() -> int:
     return _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
 
 
-def _get_x_search_retries() -> int:
-    return _get_x_search_int("retries", DEFAULT_X_SEARCH_RETRIES, 0)
-
-
-# ---------------------------------------------------------------------------
-# Credential resolution
-# ---------------------------------------------------------------------------
-
 def _resolve_xai_bearer() -> Tuple[str, str, str]:
     """Return ``(api_key, base_url, source)``; ``source`` is ``"xai-oauth"`` or ``"xai"``.
 
-    Raises ``RuntimeError`` when no credential is usable so a credential that
-    expires between registration and invocation yields a clean tool error, not
-    a 401. ``prefer_api_key=True``: see module docstring (#88040).
+    Raises ``RuntimeError`` when no credential is usable so a credential that expires
+    between registration and invocation yields a clean tool error, not a 401.
+    ``prefer_api_key=True``: see module docstring.
     """
     creds = resolve_xai_http_credentials(prefer_api_key=True)
     api_key = str(creds.get("api_key") or "").strip()
@@ -105,23 +81,17 @@ def _resolve_xai_bearer() -> Tuple[str, str, str]:
             "to sign in with your SuperGrok subscription, or set XAI_API_KEY."
         )
     base_url = str(creds.get("base_url") or DEFAULT_XAI_BASE_URL).strip().rstrip("/")
-    source = str(creds.get("provider") or "xai")
-    return api_key, base_url, source
+    return api_key, base_url, str(creds.get("provider") or "xai")
 
 
 def check_x_search_requirements() -> bool:
     """True when xAI credentials resolve to a non-empty bearer (OAuth auto-refreshed)."""
     try:
-        creds = resolve_xai_http_credentials()
-        return bool(str(creds.get("api_key") or "").strip())
+        return bool(str(resolve_xai_http_credentials().get("api_key") or "").strip())
     except Exception:
         return False
 
 
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
 def _normalize_handles(handles: Optional[List[str]], field_name: str) -> List[str]:
     cleaned = [h for h in (str(handle or "").strip().lstrip("@") for handle in handles or []) if h]
     if len(cleaned) > MAX_HANDLES:
@@ -135,9 +105,7 @@ def _parse_iso_date(value: str, field_name: str) -> date:
     try:
         return datetime.strptime(raw, "%Y-%m-%d").date()
     except ValueError as exc:
-        raise ValueError(
-            f"{field_name} must be YYYY-MM-DD (got {raw!r})"
-        ) from exc
+        raise ValueError(f"{field_name} must be YYYY-MM-DD (got {raw!r})") from exc
 
 
 def _validate_date_range(from_date: str, to_date: str) -> None:
@@ -146,17 +114,14 @@ def _validate_date_range(from_date: str, to_date: str) -> None:
     parsed_to = _parse_iso_date(to_date, "to_date") if to_date.strip() else None
     if parsed_from and parsed_to and parsed_from > parsed_to:
         raise ValueError(
-            f"from_date ({parsed_from.isoformat()}) must be on or before "
-            f"to_date ({parsed_to.isoformat()})"
+            f"from_date ({parsed_from.isoformat()}) must be on or before to_date ({parsed_to.isoformat()})"
+        )
+    today_utc = datetime.now(timezone.utc).date()
+    if parsed_from is not None and parsed_from > today_utc:
+        raise ValueError(
+            f"from_date ({parsed_from.isoformat()}) is in the future; "
+            f"X Search only indexes past posts (today UTC is {today_utc.isoformat()})"
         )
-    if parsed_from is not None:
-        today_utc = datetime.now(timezone.utc).date()
-        if parsed_from > today_utc:
-            raise ValueError(
-                f"from_date ({parsed_from.isoformat()}) is in the future; "
-                f"X Search only indexes past posts (today UTC is "
-                f"{today_utc.isoformat()})"
-            )
 
 
 def _message_contents(payload: Dict[str, Any]):
@@ -169,14 +134,12 @@ def _extract_response_text(payload: Dict[str, Any]) -> str:
     output_text = str(payload.get("output_text") or "").strip()
     if output_text:
         return output_text
-
-    parts: List[str] = []
-    for content in _message_contents(payload):
-        if content.get("type") in {"output_text", "text"}:
-            text = str(content.get("text") or "").strip()
-            if text:
-                parts.append(text)
-    return "\n\n".join(parts).strip()
+    parts = [
+        str(content.get("text") or "").strip()
+        for content in _message_contents(payload)
+        if content.get("type") in {"output_text", "text"}
+    ]
+    return "\n\n".join(p for p in parts if p).strip()
 
 
 def _extract_inline_citations(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
@@ -197,79 +160,76 @@ def _http_error_message(exc: requests.HTTPError) -> str:
     response = getattr(exc, "response", None)
     if response is None:
         return str(exc)
-
     try:
         payload = response.json()
     except Exception:
         payload = None
-
     if isinstance(payload, dict):
         code = str(payload.get("code") or "").strip()
-        error = str(payload.get("error") or "").strip()
-        message = error or str(payload)
+        message = str(payload.get("error") or "").strip() or str(payload)
         if code and code not in message:
             message = f"{code}: {message}"
         return message or str(exc)
-
     text = str(getattr(response, "text", "") or "").strip()
-    if text:
-        return text[:500]
-    return str(exc)
+    return text[:500] if text else str(exc)
 
 
 def _error_json(error: str, exc: BaseException) -> str:
-    return json.dumps(
-        {
-            "success": False,
-            "provider": "xai",
-            "tool": "x_search",
-            "error": error,
-            "error_type": type(exc).__name__,
-        },
-        ensure_ascii=False,
-    )
+    body = {"success": False, "provider": "xai", "tool": "x_search", "error": error}
+    return json.dumps({**body, "error_type": type(exc).__name__}, ensure_ascii=False)
 
 
 def _post_with_retries(url: str, headers: Dict[str, str], payload: Dict[str, Any]) -> requests.Response:
     """POST with retries on 5xx / timeout / connection errors; re-raises the last failure."""
     timeout_seconds = _get_x_search_timeout_seconds()
-    max_retries = _get_x_search_retries()
-    response: Optional[requests.Response] = None
+    max_retries = _get_x_search_int("retries", DEFAULT_X_SEARCH_RETRIES, 0)
     for attempt in range(max_retries + 1):
         try:
             response = requests.post(url, headers=headers, json=payload, timeout=timeout_seconds)
             response.raise_for_status()
-            break
+            return response
         except requests.HTTPError as e:
             status_code = getattr(getattr(e, "response", None), "status_code", None)
             if status_code is None or status_code < 500 or attempt >= max_retries:
                 raise
-            logger.warning(
-                "x_search upstream failure on attempt %s/%s: %s",
-                attempt + 1,
-                max_retries + 1,
-                _http_error_message(e),
-            )
-            time.sleep(min(5.0, 1.5 * (attempt + 1)))
+            kind, detail = "upstream", _http_error_message(e)
         except (requests.ReadTimeout, requests.ConnectionError) as e:
             if attempt >= max_retries:
                 raise
-            logger.warning(
-                "x_search transient failure on attempt %s/%s: %s",
-                attempt + 1,
-                max_retries + 1,
-                e,
-            )
-            time.sleep(min(5.0, 1.5 * (attempt + 1)))
-
-    if response is None:
-        raise RuntimeError("x_search request did not return a response")
-    return response
+            kind, detail = "transient", e
+        logger.warning("x_search %s failure on attempt %s/%s: %s", kind, attempt + 1, max_retries + 1, detail)
+        time.sleep(min(5.0, 1.5 * (attempt + 1)))
+    raise RuntimeError("x_search request did not return a response")
 
 
-# ---------------------------------------------------------------------------
-# Tool implementation
-# ---------------------------------------------------------------------------
+def _build_x_search_tool_def(
+    allowed_x_handles, excluded_x_handles, from_date: str, to_date: str,
+    enable_image_understanding: bool, enable_video_understanding: bool,
+) -> Tuple[Dict[str, Any], List[str]]:
+    """Return ``(tool_def, active_filters)``; raises ValueError on invalid filters."""
+    allowed = _normalize_handles(allowed_x_handles, "allowed_x_handles")
+    excluded = _normalize_handles(excluded_x_handles, "excluded_x_handles")
+    if allowed and excluded:
+        raise ValueError("allowed_x_handles and excluded_x_handles cannot be used together")
+    _validate_date_range(from_date, to_date)
+
+    tool_def: Dict[str, Any] = {"type": "x_search"}
+    active_filters: List[str] = []
+    for key, value in (
+        ("allowed_x_handles", allowed),
+        ("excluded_x_handles", excluded),
+        ("from_date", from_date.strip()),
+        ("to_date", to_date.strip()),
+    ):
+        if value:
+            tool_def[key] = value
+            active_filters.append(key)
+    if enable_image_understanding:
+        tool_def["enable_image_understanding"] = True
+    if enable_video_understanding:
+        tool_def["enable_video_understanding"] = True
+    return tool_def, active_filters
+
 
 def x_search_tool(
     query: str,
@@ -282,43 +242,23 @@ def x_search_tool(
 ) -> str:
     if not query or not query.strip():
         return tool_error("query is required for x_search")
-
     try:
         api_key, base_url, source = _resolve_xai_bearer()
     except RuntimeError as exc:
         return tool_error(str(exc))
 
     try:
-        allowed = _normalize_handles(allowed_x_handles, "allowed_x_handles")
-        excluded = _normalize_handles(excluded_x_handles, "excluded_x_handles")
-        if allowed and excluded:
-            return tool_error("allowed_x_handles and excluded_x_handles cannot be used together")
-
         try:
-            _validate_date_range(from_date, to_date)
+            tool_def, active_filters = _build_x_search_tool_def(
+                allowed_x_handles, excluded_x_handles, from_date, to_date,
+                enable_image_understanding, enable_video_understanding,
+            )
             reasoning_effort = _get_x_search_reasoning_effort()
         except ValueError as exc:
             return tool_error(str(exc))
 
-        from_date, to_date = from_date.strip(), to_date.strip()
-        tool_def: Dict[str, Any] = {"type": "x_search"}
-        active_filters: List[str] = []
-        for key, value in (
-            ("allowed_x_handles", allowed),
-            ("excluded_x_handles", excluded),
-            ("from_date", from_date),
-            ("to_date", to_date),
-        ):
-            if value:
-                tool_def[key] = value
-                active_filters.append(key)
-        if enable_image_understanding:
-            tool_def["enable_image_understanding"] = True
-        if enable_video_understanding:
-            tool_def["enable_video_understanding"] = True
-
         payload = {
-            "model": _get_x_search_model(),
+            "model": str(_load_x_search_config().get("model") or "").strip() or DEFAULT_X_SEARCH_MODEL,
             "input": [{"role": "user", "content": query.strip()}],
             "tools": [tool_def],
             "store": False,
@@ -326,22 +266,17 @@ def x_search_tool(
         if reasoning_effort:
             payload["reasoning"] = {"effort": reasoning_effort}
 
-        response = _post_with_retries(
-            f"{base_url}/responses",
-            {
-                "Authorization": f"Bearer {api_key}",
-                "Content-Type": "application/json",
-                "User-Agent": hermes_xai_user_agent(),
-            },
-            payload,
-        )
-        data = response.json()
+        headers = {
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+            "User-Agent": hermes_xai_user_agent(),
+        }
+        data = _post_with_retries(f"{base_url}/responses", headers, payload).json()
 
         citations = list(data.get("citations") or [])
         inline_citations = _extract_inline_citations(data)
-        # xAI returns 200 with a synthesized answer even when no posts match the
-        # narrowing filters; with both citation channels empty the answer came
-        # from training data, so flag it as degraded.
+        # xAI returns 200 with a synthesized answer even when no posts match the narrowing
+        # filters; with both citation channels empty the answer came from training data.
         degraded = bool(active_filters) and not citations and not inline_citations
         return json.dumps(
             {
@@ -356,9 +291,7 @@ def x_search_tool(
                 "inline_citations": inline_citations,
                 "degraded": degraded,
                 "degraded_reason": (
-                    f"no citations returned despite filters: {', '.join(active_filters)}"
-                    if degraded
-                    else None
+                    f"no citations returned despite filters: {', '.join(active_filters)}" if degraded else None
                 ),
             },
             ensure_ascii=False,
@@ -428,23 +361,15 @@ X_SEARCH_SCHEMA = {
 
 def _handle_x_search(args, **kw):
     return x_search_tool(
-        query=args.get("query", ""),
-        allowed_x_handles=args.get("allowed_x_handles"),
-        excluded_x_handles=args.get("excluded_x_handles"),
-        from_date=args.get("from_date", ""),
-        to_date=args.get("to_date", ""),
-        enable_image_understanding=bool(args.get("enable_image_understanding", False)),
-        enable_video_understanding=bool(args.get("enable_video_understanding", False)),
+        args.get("query", ""), args.get("allowed_x_handles"), args.get("excluded_x_handles"),
+        args.get("from_date", ""), args.get("to_date", ""),
+        bool(args.get("enable_image_understanding", False)),
+        bool(args.get("enable_video_understanding", False)),
     )
 
 
 registry.register(
-    name="x_search",
-    toolset="x_search",
-    schema=X_SEARCH_SCHEMA,
-    handler=_handle_x_search,
-    check_fn=check_x_search_requirements,
-    requires_env=["XAI_API_KEY"],
-    emoji="🐦",
+    name="x_search", toolset="x_search", schema=X_SEARCH_SCHEMA, handler=_handle_x_search,
+    check_fn=check_x_search_requirements, requires_env=["XAI_API_KEY"], emoji="🐦",
     max_result_size_chars=100_000,
 )
diff --git a/tools/yuanbao_tools.py b/tools/yuanbao_tools.py
index b7b26407b1..928c3ae735 100644
--- a/tools/yuanbao_tools.py
+++ b/tools/yuanbao_tools.py
@@ -142,7 +142,8 @@ async def query_group_members(args) -> dict:
 
     action: find (按昵称模糊搜索; 无 name 时等同 list_all) / list_bots / list_all (默认).
     """
-    group_code, action, name = args.get("group_code", ""), args.get("action", "list_all"), args.get("name", "")
+    group_code, name = args.get("group_code", ""), args.get("name", "")
+    action = args.get("action", "list_all")
     if not group_code:
         return _err("group_code is required")
     all_members = [

From 92f8a30ffd8e09efeb9d69b29896caaf6ac3c661 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:19:15 -0700
Subject: [PATCH 3/5] refactor(tools): xai_http compact defensive branches and
 docstrings

---
 tools/xai_http.py | 103 ++++++++++++++++++----------------------------
 1 file changed, 39 insertions(+), 64 deletions(-)

diff --git a/tools/xai_http.py b/tools/xai_http.py
index 33cced4d03..dd0fa38d47 100644
--- a/tools/xai_http.py
+++ b/tools/xai_http.py
@@ -21,14 +21,12 @@ def _dict_get(obj: Any, key: str) -> Any:
 def has_xai_credentials() -> bool:
     """Cheap probe: True when xAI credentials are *likely* usable.
 
-    Deliberately avoids :func:`resolve_xai_http_credentials` (disk locks, OAuth
-    network refresh) — hot-paint callers and ``WebSearchProvider.is_available()``
-    must not do network I/O. Checks, fast-to-slow: ``XAI_API_KEY``; a non-empty
-    ``providers.xai-oauth.tokens.access_token`` in ``auth.json``; any
-    ``credential_pool.xai-oauth`` entry with an ``access_token`` (pool-only
-    multi-account grants never write the providers singleton). Returns False on
-    any exception so a corrupted auth store can't block other availability
-    scans; truthful refresh/expiry handling happens in the caller's request.
+    Deliberately avoids :func:`resolve_xai_http_credentials` (disk locks, OAuth network
+    refresh) — hot-paint callers and ``WebSearchProvider.is_available()`` must not do network
+    I/O. Checks, fast-to-slow: ``XAI_API_KEY``; ``providers.xai-oauth.tokens.access_token`` in
+    ``auth.json``; any ``credential_pool.xai-oauth`` entry with an ``access_token`` (pool-only
+    multi-account grants never write the providers singleton). Returns False on any exception
+    so a corrupted auth store can't block other availability scans.
     """
     try:
         from agent.secret_scope import get_secret
@@ -49,11 +47,9 @@ def has_xai_credentials() -> bool:
         if str(_dict_get(tokens, "access_token") or "").strip():
             return True
         entries = _dict_get(_dict_get(store, "credential_pool"), "xai-oauth")
-        if isinstance(entries, list):
-            for entry in entries:
-                if isinstance(entry, dict) and str(entry.get("access_token", "") or "").strip():
-                    return True
-        return False
+        return isinstance(entries, list) and any(
+            isinstance(e, dict) and str(e.get("access_token", "") or "").strip() for e in entries
+        )
     except Exception:
         return False
 
@@ -61,14 +57,12 @@ def has_xai_credentials() -> bool:
 def get_env_value(name: str, default=None):
     """Read ``name`` from ``~/.hermes/.env`` first, then ``os.environ``.
 
-    Wraps :func:`hermes_cli.config.get_env_value` so tests can patch
-    ``tools.xai_http.get_env_value`` to inject dotenv-only secrets.
+    Wraps :func:`hermes_cli.config.get_env_value` so tests can patch ``tools.xai_http.get_env_value``.
     """
     try:
         from hermes_cli.config import get_env_value as _hermes_get_env_value
     except ImportError:
         return os.environ.get(name, default)
-
     value = _hermes_get_env_value(name)
     return value if value is not None else default
 
@@ -98,15 +92,17 @@ def _load_config_section(section_name: str) -> Dict[str, Any]:
         return {}
 
 
+_TRUE_WORDS = {"1", "true", "yes", "on", "enabled"}
+_FALSE_WORDS = {"0", "false", "no", "off", "disabled"}
+
+
 def _coerce_bool(value: Any, default: bool) -> bool:
     if isinstance(value, bool):
         return value
     if isinstance(value, str):
         normalized = value.strip().lower()
-        if normalized in {"1", "true", "yes", "on", "enabled"}:
-            return True
-        if normalized in {"0", "false", "no", "off", "disabled"}:
-            return False
+        if normalized in _TRUE_WORDS | _FALSE_WORDS:
+            return normalized in _TRUE_WORDS
     return default
 
 
@@ -124,18 +120,15 @@ def _coerce_expires_after(value: Any) -> Optional[int]:
             return SAFE_XAI_STORAGE_EXPIRES_AFTER_SECONDS
     if isinstance(value, (int, float)):
         seconds = int(value)
-        if seconds <= 0:
-            return None
-        return min(seconds, MAX_XAI_STORAGE_EXPIRES_AFTER_SECONDS)
+        return None if seconds <= 0 else min(seconds, MAX_XAI_STORAGE_EXPIRES_AFTER_SECONDS)
     return SAFE_XAI_STORAGE_EXPIRES_AFTER_SECONDS
 
 
 def read_xai_imagine_storage_config(section_name: str) -> Dict[str, Any]:
-    """Read ``<section_name>.xai.storage`` (``image_gen`` or ``video_gen``).
+    """Read ``<section_name>.xai.storage`` (``image_gen``/``video_gen``) -> {enabled, public_url, expires_after}.
 
-    Shape: ``storage: {enabled: true, public_url: true, expires_after: null}``.
-    Storage is on by default so xAI returns permanent public URLs instead of
-    short-lived CDN URLs; ``expires_after: null`` means permanent.
+    Storage is on by default so xAI returns permanent public URLs instead of short-lived CDN
+    URLs; ``expires_after: null`` means permanent.
     """
     storage = _dict_get(_dict_get(_load_config_section(section_name), "xai"), "storage")
     storage = storage if isinstance(storage, dict) else {}
@@ -147,22 +140,17 @@ def read_xai_imagine_storage_config(section_name: str) -> Dict[str, Any]:
 
 
 def build_xai_storage_options(
-    section_name: str,
-    *,
-    filename_prefix: str,
-    extension: str,
+    section_name: str, *, filename_prefix: str, extension: str,
 ) -> Optional[Dict[str, Any]]:
     """Return an xAI ``storage_options`` payload, or None when disabled."""
     cfg = read_xai_imagine_storage_config(section_name)
     if not cfg["enabled"]:
         return None
-
     ts = datetime.datetime.now(datetime.UTC).strftime("%Y%m%d-%H%M%S")
     short = uuid.uuid4().hex[:8]
     ext = extension.lstrip(".") or "bin"
     payload: Dict[str, Any] = {
-        "filename": f"{filename_prefix}-{ts}-{short}.{ext}",
-        "public_url": bool(cfg["public_url"]),
+        "filename": f"{filename_prefix}-{ts}-{short}.{ext}", "public_url": bool(cfg["public_url"]),
     }
     if cfg["expires_after"] is not None:
         payload["expires_after"] = cfg["expires_after"]
@@ -174,9 +162,8 @@ def xai_storage_notice_text(section_name: str) -> str:
     cfg = read_xai_imagine_storage_config(section_name)
     if not cfg["enabled"]:
         return ""
-    if cfg["expires_after"] is None:
-        retention = "without an automatic expiry"
-    else:
+    retention = "without an automatic expiry"
+    if cfg["expires_after"] is not None:
         days = cfg["expires_after"] / (24 * 60 * 60)
         retention = f"for about {days:g} day{'s' if days != 1 else ''}"
     return (
@@ -209,9 +196,8 @@ def maybe_mark_xai_storage_notice_seen(section_name: str) -> Optional[str]:
 def _resolve_explicit_xai_api_key() -> str:
     """Read ``XAI_API_KEY`` via ``resolve_provider_secret`` (config → profile scope → env/.env → pool).
 
-    Both the preferred-key and the no-OAuth fallback paths go through here so
-    scope policy (incl. failing closed in a multiplexed gateway turn) is never
-    re-implemented per caller.
+    Both the preferred-key and the no-OAuth fallback paths go through here so scope policy
+    (incl. failing closed in a multiplexed gateway turn) is never re-implemented per caller.
     """
     try:
         from tools.tool_backend_helpers import resolve_provider_secret
@@ -227,12 +213,8 @@ def _xai_base_url_override() -> str:
 
 
 def _resolve_explicit_xai_base_url(default: str = DEFAULT_XAI_BASE_URL) -> str:
-    """Base URL for the explicit-API-key path.
-
-    Origin-pinned via ``_xai_validate_inference_base_url`` so a tampered env
-    override can't exfiltrate the bearer; on rejection falls back to ``default``
-    rather than raising.
-    """
+    """Base URL for the explicit-API-key path, origin-pinned via ``_xai_validate_inference_base_url``
+    so a tampered env override can't exfiltrate the bearer; rejection falls back to ``default``."""
     override = _xai_base_url_override()
     try:
         import hermes_cli.auth as auth_mod
@@ -243,21 +225,17 @@ def _resolve_explicit_xai_base_url(default: str = DEFAULT_XAI_BASE_URL) -> str:
 
 
 def resolve_xai_http_credentials(
-    *,
-    force_refresh: bool = False,
-    api_key_hint: Optional[str] = None,
-    prefer_api_key: bool = False,
+    *, force_refresh: bool = False, api_key_hint: Optional[str] = None, prefer_api_key: bool = False,
 ) -> Dict[str, str]:
     """Resolve bearer credentials for direct xAI HTTP endpoints.
 
-    Default order: Hermes-managed xAI OAuth, then ``XAI_API_KEY`` (read via
-    ``get_env_value`` so ``~/.hermes/.env`` keys count). ``prefer_api_key=True``
-    inverts that for API-metered endpoints where the subscription OAuth bearer
-    authorizes but misbehaves (x_search answers without citations, TTS 403s).
-    Both branches honor ``HERMES_XAI_BASE_URL``/``XAI_BASE_URL`` behind the same
-    origin-pinning validation. ``force_refresh=True`` forces an OAuth refresh;
-    pass the rejected bearer as ``api_key_hint`` so a multi-account pool
-    refreshes the issuing entry, not whichever its strategy selects first.
+    Default order: Hermes-managed xAI OAuth, then ``XAI_API_KEY`` (via ``get_env_value`` so
+    ``~/.hermes/.env`` keys count). ``prefer_api_key=True`` inverts that for API-metered
+    endpoints where the subscription OAuth bearer authorizes but misbehaves (x_search answers
+    without citations, TTS 403s). Both branches honor ``HERMES_XAI_BASE_URL``/``XAI_BASE_URL``
+    behind the same origin-pinning validation. ``force_refresh=True`` forces an OAuth refresh;
+    pass the rejected bearer as ``api_key_hint`` so a multi-account pool refreshes the issuing
+    entry, not whichever its strategy selects first.
     """
     if prefer_api_key:
         explicit_key = str(_resolve_explicit_xai_api_key() or "").strip()
@@ -275,18 +253,15 @@ def resolve_xai_http_credentials(
             # the next healthy account rather than resurrecting the stale row.
             entry = pool.select()
         access_token = str(
-            getattr(entry, "runtime_api_key", None)
-            or getattr(entry, "access_token", "")
+            getattr(entry, "runtime_api_key", None) or getattr(entry, "access_token", "")
         ).strip()
         fallback_base_url = str(
             getattr(entry, "runtime_base_url", None)
             or getattr(entry, "base_url", "")
             or auth_mod.DEFAULT_XAI_OAUTH_BASE_URL
         ).strip().rstrip("/")
-        base_url = auth_mod._xai_validate_inference_base_url(
-            _xai_base_url_override(),
-            fallback=fallback_base_url,
-        )
+        override = _xai_base_url_override()
+        base_url = auth_mod._xai_validate_inference_base_url(override, fallback=fallback_base_url)
         if access_token:
             return {"provider": "xai-oauth", "api_key": access_token, "base_url": base_url}
     except Exception:

From fc805419eee09a5f21338df2036d4504ef13a8ea Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:21:59 -0700
Subject: [PATCH 4/5] refactor(tools): xai_video_tools compact helpers and
 registration

---
 tools/xai_video_tools.py | 40 ++++++++++------------------------------
 1 file changed, 10 insertions(+), 30 deletions(-)

diff --git a/tools/xai_video_tools.py b/tools/xai_video_tools.py
index 2e7647fde5..fed8230838 100644
--- a/tools/xai_video_tools.py
+++ b/tools/xai_video_tools.py
@@ -7,11 +7,7 @@ import json
 from typing import Any, Dict, Optional
 
 from hermes_cli.config import load_config
-from plugins.video_gen.xai import (
-    has_xai_video_credentials,
-    run_xai_video_edit,
-    run_xai_video_extend,
-)
+from plugins.video_gen.xai import has_xai_video_credentials, run_xai_video_edit, run_xai_video_extend
 from tools.registry import registry, tool_error
 
 
@@ -29,14 +25,11 @@ def _check_xai_video_requirements() -> bool:
 
 
 def _clean_string(value: Any) -> Optional[str]:
-    if isinstance(value, str) and value.strip():
-        return value.strip()
-    return None
+    return value.strip() if isinstance(value, str) and value.strip() else None
 
 
 def _coerce_int(value: Any) -> Optional[int]:
-    # bool is rejected (unlike video_generation_tool._coerce_int) so
-    # duration=true never becomes 1.
+    # bool is rejected (unlike video_generation_tool._coerce_int) so duration=true never becomes 1.
     if value is None or isinstance(value, bool):
         return None
     try:
@@ -48,9 +41,7 @@ def _coerce_int(value: Any) -> Optional[int]:
 def _normalize_public_video_url(video_url: Any) -> Optional[str]:
     """Require a public HTTPS MP4 URL (``http``/``https`` only)."""
     cleaned = _clean_string(video_url)
-    if cleaned and cleaned.lower().startswith(("http://", "https://")):
-        return cleaned
-    return None
+    return cleaned if cleaned and cleaned.lower().startswith(("http://", "https://")) else None
 
 
 _VIDEO_URL_PARAM = {
@@ -60,10 +51,7 @@ _VIDEO_URL_PARAM = {
         "`public_url` from a prior xAI Imagine result."
     ),
 }
-_MODEL_PARAM = {
-    "type": "string",
-    "description": "Optional xAI Imagine model override.",
-}
+_MODEL_PARAM = {"type": "string", "description": "Optional xAI Imagine model override."}
 
 
 def _xai_video_schema(name: str, verb: str, noun: str, prompt_verb: str, extra: Dict[str, Any]) -> Dict[str, Any]:
@@ -111,8 +99,6 @@ 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"))
-    model = _clean_string(args.get("model"))
-
     if not prompt:
         return tool_error(f"prompt is required for xAI video {op}")
     if not video_url:
@@ -131,6 +117,7 @@ def _run_xai_video_tool(args: Dict[str, Any], op: str, run, **extra: Any) -> str
             "provider": "xai",
         })
 
+    model = _clean_string(args.get("model"))
     return json.dumps(run(prompt=prompt, video_url=video_url, model=model, **extra))
 
 
@@ -139,9 +126,8 @@ def _handle_xai_video_edit(args: Dict[str, Any], **_kw: Any) -> str:
 
 
 def _handle_xai_video_extend(args: Dict[str, Any], **_kw: Any) -> str:
-    return _run_xai_video_tool(
-        args, "extend", run_xai_video_extend, duration=_coerce_int(args.get("duration")),
-    )
+    duration = _coerce_int(args.get("duration"))
+    return _run_xai_video_tool(args, "extend", run_xai_video_extend, duration=duration)
 
 
 for _name, _schema, _handler in (
@@ -149,12 +135,6 @@ for _name, _schema, _handler in (
     ("xai_video_extend", XAI_VIDEO_EXTEND_SCHEMA, _handle_xai_video_extend),
 ):
     registry.register(
-        name=_name,
-        toolset="video_gen",
-        schema=_schema,
-        handler=_handler,
-        check_fn=_check_xai_video_requirements,
-        requires_env=[],
-        is_async=False,
-        emoji="video",
+        name=_name, toolset="video_gen", schema=_schema, handler=_handler,
+        check_fn=_check_xai_video_requirements, requires_env=[], is_async=False, emoji="video",
     )

From 54df4e53d469d315bde27059be62b8dd714dfccb Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Wed, 2 Sep 2026 22:27:29 -0700
Subject: [PATCH 5/5] refactor(tools): x_search/yuanbao final docstring
 compaction and blank-date parse merge

---
 tools/x_search_tool.py | 42 ++++++++++++++++--------------------------
 tools/yuanbao_tools.py | 22 +++++++---------------
 2 files changed, 23 insertions(+), 41 deletions(-)

diff --git a/tools/x_search_tool.py b/tools/x_search_tool.py
index fabce55fe1..070af49031 100644
--- a/tools/x_search_tool.py
+++ b/tools/x_search_tool.py
@@ -1,16 +1,12 @@
 #!/usr/bin/env python3
 """X Search tool backed by xAI's built-in ``x_search`` Responses API tool.
 
-Registers when either xAI credential path is available (``XAI_API_KEY`` or
-``hermes auth add xai-oauth``). At call time an explicit ``XAI_API_KEY`` wins
-(``prefer_api_key=True``): x_search is API-metered and the subscription OAuth
-bearer answers ``/v1/responses`` in a degraded no-citation mode.
-
-``from_date``/``to_date`` are validated client-side (strict ``YYYY-MM-DD``, ``from <= to``,
-``from`` not in the future) so malformed windows fail fast instead of burning a billable
-call. Successful responses carry ``degraded``/``degraded_reason``: True when a narrowing
-filter was active AND xAI returned no citations in either channel (answer came from the
-model's own knowledge, not the X index).
+Registers when either xAI credential path is available (``XAI_API_KEY`` or ``hermes auth add
+xai-oauth``). At call time an explicit ``XAI_API_KEY`` wins (``prefer_api_key=True``): x_search
+is API-metered and the subscription OAuth bearer answers ``/v1/responses`` without citations.
+Date filters are validated client-side so malformed windows fail fast instead of burning a
+billable call. Results carry ``degraded``: True when a narrowing filter was active AND xAI
+returned no citations in either channel (answer came from model knowledge, not the X index).
 """
 
 from __future__ import annotations
@@ -62,16 +58,11 @@ def _get_x_search_int(key: str, default: int, floor: int) -> int:
         return default
 
 
-def _get_x_search_timeout_seconds() -> int:
-    return _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
-
-
 def _resolve_xai_bearer() -> Tuple[str, str, str]:
     """Return ``(api_key, base_url, source)``; ``source`` is ``"xai-oauth"`` or ``"xai"``.
 
-    Raises ``RuntimeError`` when no credential is usable so a credential that expires
-    between registration and invocation yields a clean tool error, not a 401.
-    ``prefer_api_key=True``: see module docstring.
+    Raises ``RuntimeError`` when no credential is usable so a credential that expires between
+    registration and invocation yields a clean tool error, not a 401.
     """
     creds = resolve_xai_http_credentials(prefer_api_key=True)
     api_key = str(creds.get("api_key") or "").strip()
@@ -99,9 +90,11 @@ def _normalize_handles(handles: Optional[List[str]], field_name: str) -> List[st
     return cleaned
 
 
-def _parse_iso_date(value: str, field_name: str) -> date:
-    """Parse a strict YYYY-MM-DD string (xAI silently accepts malformed dates and returns no citations)."""
+def _parse_iso_date(value: str, field_name: str) -> Optional[date]:
+    """Strict YYYY-MM-DD or None for blank (xAI silently accepts malformed dates and returns no citations)."""
     raw = value.strip()
+    if not raw:
+        return None
     try:
         return datetime.strptime(raw, "%Y-%m-%d").date()
     except ValueError as exc:
@@ -110,8 +103,7 @@ def _parse_iso_date(value: str, field_name: str) -> date:
 
 def _validate_date_range(from_date: str, to_date: str) -> None:
     """Both parse as YYYY-MM-DD; from <= to; from not after today UTC (to may be in the future)."""
-    parsed_from = _parse_iso_date(from_date, "from_date") if from_date.strip() else None
-    parsed_to = _parse_iso_date(to_date, "to_date") if to_date.strip() else None
+    parsed_from, parsed_to = _parse_iso_date(from_date, "from_date"), _parse_iso_date(to_date, "to_date")
     if parsed_from and parsed_to and parsed_from > parsed_to:
         raise ValueError(
             f"from_date ({parsed_from.isoformat()}) must be on or before to_date ({parsed_to.isoformat()})"
@@ -181,7 +173,7 @@ def _error_json(error: str, exc: BaseException) -> str:
 
 def _post_with_retries(url: str, headers: Dict[str, str], payload: Dict[str, Any]) -> requests.Response:
     """POST with retries on 5xx / timeout / connection errors; re-raises the last failure."""
-    timeout_seconds = _get_x_search_timeout_seconds()
+    timeout_seconds = _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
     max_retries = _get_x_search_int("retries", DEFAULT_X_SEARCH_RETRIES, 0)
     for attempt in range(max_retries + 1):
         try:
@@ -256,7 +248,6 @@ def x_search_tool(
             reasoning_effort = _get_x_search_reasoning_effort()
         except ValueError as exc:
             return tool_error(str(exc))
-
         payload = {
             "model": str(_load_x_search_config().get("model") or "").strip() or DEFAULT_X_SEARCH_MODEL,
             "input": [{"role": "user", "content": query.strip()}],
@@ -265,14 +256,12 @@ def x_search_tool(
         }
         if reasoning_effort:
             payload["reasoning"] = {"effort": reasoning_effort}
-
         headers = {
             "Authorization": f"Bearer {api_key}",
             "Content-Type": "application/json",
             "User-Agent": hermes_xai_user_agent(),
         }
         data = _post_with_retries(f"{base_url}/responses", headers, payload).json()
-
         citations = list(data.get("citations") or [])
         inline_citations = _extract_inline_citations(data)
         # xAI returns 200 with a synthesized answer even when no posts match the narrowing
@@ -301,7 +290,8 @@ def x_search_tool(
         return _error_json(_http_error_message(e), e)
     except requests.ReadTimeout as e:
         logger.error("x_search timed out: %s", e, exc_info=True)
-        return _error_json(f"xAI x_search timed out after {_get_x_search_timeout_seconds()} seconds", e)
+        timeout = _get_x_search_int("timeout_seconds", DEFAULT_X_SEARCH_TIMEOUT_SECONDS, 30)
+        return _error_json(f"xAI x_search timed out after {timeout} seconds", e)
     except Exception as e:
         logger.error("x_search failed: %s", e, exc_info=True)
         return _error_json(str(e), e)
diff --git a/tools/yuanbao_tools.py b/tools/yuanbao_tools.py
index 928c3ae735..80c5a7db99 100644
--- a/tools/yuanbao_tools.py
+++ b/tools/yuanbao_tools.py
@@ -41,8 +41,7 @@ def _err(msg: str) -> dict:
 
 
 def _yb_tool(label: str):
-    """Handler decorator: ``fn(args)`` result → tool_result; ``_YbError`` → its envelope;
-    any other exception → logged + ``_err(str(exc))``."""
+    """Handler decorator: ``fn(args)`` → tool_result; ``_YbError`` → its envelope; else logged ``_err``."""
     def deco(fn):
         @functools.wraps(fn)
         async def handler(args, **kw):
@@ -242,12 +241,10 @@ async def send_sticker(args) -> dict:
 async def send_dm(args) -> dict:
     """Send a DM to a group member, with optional media.
 
-    group_code: explicit arg, else the session's "group:<code>" chat_id. Without ``user_id``
-    the member list is searched by ``name`` (partial, case-insensitive; >1 match returns
-    candidates). media_files items are {"path", "is_voice"} dicts or (path, is_voice) pairs;
-    ``MEDIA:<path>`` tags embedded in the text are extracted too. Text goes via adapter.send_dm;
-    media via send_image_file (images) or send_document. Partial media failures are reported
-    in ``note``, not as failure.
+    group_code defaults to the session's "group:<code>" chat_id. Without ``user_id`` the member
+    list is searched by ``name`` (partial, case-insensitive; >1 match returns candidates).
+    media_files items are {"path", "is_voice"} dicts or (path, is_voice) pairs; ``MEDIA:<path>``
+    tags in the text count too. Partial media failures are reported in ``note``, not as failure.
     """
     group_code = args.get("group_code", "")
     if not group_code:
@@ -530,11 +527,6 @@ _TOOLS = (
 
 for _schema, _handler, _emoji in _TOOLS:
     registry.register(
-        name=_schema["name"],
-        toolset="hermes-yuanbao",
-        schema=_schema,
-        handler=_handler,
-        check_fn=_check_yuanbao,
-        is_async=True,
-        emoji=_emoji,
+        name=_schema["name"], toolset="hermes-yuanbao", schema=_schema, handler=_handler,
+        check_fn=_check_yuanbao, is_async=True, emoji=_emoji,
     )