fix(loop): strip input_audio and video_url before session persistence

Code review found that _sanitize_persisted_blocks only stripped
image_url blocks, causing base64-encoded audio/video payloads to
bloat session history files.

- Extend _sanitize_persisted_blocks to replace input_audio and
  video_url blocks with text placeholders using LLMProvider._media_placeholder.
- Add tests for audio/video stripping with and without _meta.path.

All 3267 tests pass.

.gitignore

@@ -97,4 +97,3 @@ logs/
 tmp/
 temp/
 *.tmp
-exp/

docs/configuration.md

@@ -168,7 +168,7 @@ ANTHROPIC_API_KEY="$(bw get password api/anthropic)" nanobot agent
 <details>
 <summary><b>Skywork / APIFree</b></summary>
 
-Skywork uses APIFree's OpenAI-compatible Agent API endpoint. Configure the provider
+Skywork uses the OpenAI-compatible APIFree API endpoint. Configure the provider
 once, then use Skywork model IDs such as `skywork-ai/skyclaw-v1`.
 
 ```json
@@ -176,7 +176,7 @@ once, then use Skywork model IDs such as `skywork-ai/skyclaw-v1`.
   "providers": {
     "skywork": {
       "apiKey": "${SKYWORK_API_KEY}",
-      "apiBase": "https://api.apifree.ai/agent/v1"
+      "apiBase": "https://api.apifree.ai/v1"
     }
   },
   "agents": {

nanobot/__init__.py

@@ -2,10 +2,9 @@
 nanobot - A lightweight AI agent framework
 """
 
-import tomllib
-from importlib.metadata import PackageNotFoundError
-from importlib.metadata import version as _pkg_version
+from importlib.metadata import PackageNotFoundError, version as _pkg_version
 from pathlib import Path
+import tomllib
 
 
 def _read_pyproject_version() -> str | None:
@@ -28,21 +27,6 @@ def _resolve_version() -> str:
 __version__ = _resolve_version()
 __logo__ = "🐈"
 
-_LAZY_EXPORTS = {
-    "Nanobot": ".nanobot",
-    "RunResult": ".nanobot",
-}
-
-
-def __getattr__(name: str):
-    module_path = _LAZY_EXPORTS.get(name)
-    if module_path is None:
-        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-    from importlib import import_module
-    mod = import_module(module_path, __name__)
-    val = getattr(mod, name)
-    globals()[name] = val
-    return val
-
+from nanobot.nanobot import Nanobot, RunResult
 
 __all__ = ["Nanobot", "RunResult"]

nanobot/agent/context.py

@@ -2,6 +2,7 @@
 
 import base64
 import mimetypes
+import os
 import platform
 from contextlib import suppress
 from importlib.resources import files as pkg_files
@@ -10,11 +11,16 @@ from typing import Any, Mapping, Sequence
 
 from nanobot.agent.memory import MemoryStore
 from nanobot.agent.skills import SkillsLoader
+from nanobot.config.schema import InputLimitsConfig
 from nanobot.session.goal_state import goal_state_runtime_lines
 from nanobot.utils.helpers import (
+    audio_format_for_api,
+    audio_mime_compat,
     current_time_str,
+    detect_audio_mime,
     detect_image_mime,
     truncate_text,
+    video_mime_compat,
 )
 from nanobot.utils.prompt_templates import render_template
 
@@ -28,11 +34,12 @@ class ContextBuilder:
     _MAX_HISTORY_CHARS = 32_000  # hard cap on recent history section size
     _RUNTIME_CONTEXT_END = "[/Runtime Context]"
 
-    def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None):
+    def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | None = None, input_limits: InputLimitsConfig | None = None):
         self.workspace = workspace
         self.timezone = timezone
         self.memory = MemoryStore(workspace)
         self.skills = SkillsLoader(workspace, disabled_skills=set(disabled_skills) if disabled_skills else None)
+        self.input_limits = input_limits or InputLimitsConfig()
 
     def build_system_prompt(
         self,
@@ -142,6 +149,28 @@ class ContextBuilder:
                 return content.strip() == tpl.read_text(encoding="utf-8").strip()
         return False
 
+    @staticmethod
+    def _file_size_ok(p: Path, max_bytes: int) -> bool | None:
+        """Check file size via stat without reading into memory.
+
+        Returns True if size is within limit, False if oversized,
+        None if file cannot be stat'd (caller should try read_bytes instead).
+        """
+        try:
+            return os.stat(p).st_size <= max_bytes
+        except OSError:
+            return None
+
+    @staticmethod
+    def _encode_image_block(raw: bytes, mime: str, path: Path) -> dict[str, Any]:
+        """Base64-encode file bytes into an image_url content block."""
+        b64 = base64.b64encode(raw).decode()
+        return {
+            "type": "image_url",
+            "image_url": {"url": f"data:{mime};base64,{b64}"},
+            "_meta": {"path": str(path)},
+        }
+
     def build_messages(
         self,
         history: list[dict[str, Any]],
@@ -154,6 +183,9 @@ class ContextBuilder:
         sender_id: str | None = None,
         session_summary: str | None = None,
         session_metadata: Mapping[str, Any] | None = None,
+        supports_vision: bool | None = None,
+        supports_audio: bool | None = None,
+        supports_video: bool | None = None,
     ) -> list[dict[str, Any]]:
         """Build the complete message list for an LLM call."""
         extra = goal_state_runtime_lines(session_metadata)
@@ -164,7 +196,12 @@ class ContextBuilder:
             sender_id=sender_id,
             supplemental_lines=extra or None,
         )
-        user_content = self._build_user_content(current_message, media)
+        user_content = self._build_user_content(
+            current_message, media,
+            supports_vision=supports_vision,
+            supports_audio=supports_audio,
+            supports_video=supports_video,
+        )
 
         # Merge runtime context and user content into a single user message
         # to avoid consecutive same-role messages that some providers reject.
@@ -186,28 +223,171 @@ class ContextBuilder:
         messages.append({"role": current_role, "content": merged})
         return messages
 
-    def _build_user_content(self, text: str, media: list[str] | None) -> str | list[dict[str, Any]]:
-        """Build user message content with optional base64-encoded images."""
+    def _build_user_content(
+        self,
+        text: str,
+        media: list[str] | None,
+        *,
+        supports_vision: bool | None = None,
+        supports_audio: bool | None = None,
+        supports_video: bool | None = None,
+    ) -> str | list[dict[str, Any]]:
+        """Build user message content with optional media blocks.
+
+        Args:
+            text: The user text message.
+            media: List of file paths to media files.
+            supports_vision: True=model supports images, False=use placeholder,
+                             None=unconfigured (send images as before, let
+                             provider/retry handle degradation).
+            supports_audio: True=model supports native audio, False/None=skip
+                            (channel layer already transcribed).
+            supports_video: True=model supports native video, False/None=use
+                            [file: path] placeholder.
+        """
         if not media:
             return text
 
-        images = []
+        blocks: list[dict[str, Any]] = []
+        notes: list[str] = []
+        limits = self.input_limits
+
+        # Enforce image count limit
+        max_images = limits.max_input_images
+        image_count = 0
+        image_media = []
+        non_image_media = []
         for path in media:
+            p = Path(path)
+            guessed_mime = mimetypes.guess_type(path)[0] or ""
+            if guessed_mime.startswith("image/"):
+                image_count += 1
+                if image_count <= max_images:
+                    image_media.append(path)
+            else:
+                non_image_media.append(path)
+
+        if image_count > max_images:
+            extra = image_count - max_images
+            noun = "image" if extra == 1 else "images"
+            notes.append(
+                f"[Skipped {extra} {noun}: "
+                f"only the first {max_images} images are included]"
+            )
+
+        # Process images
+        for path in image_media:
             p = Path(path)
             if not p.is_file():
                 continue
-            raw = p.read_bytes()
-            mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]
-            if not mime or not mime.startswith("image/"):
+
+            # When explicitly marked as non-vision, downgrade to text placeholder
+            if supports_vision is False:
+                blocks.append({"type": "text", "text": f"[image: {p}]"})
                 continue
-            b64 = base64.b64encode(raw).decode()
-            images.append({
-                "type": "image_url",
-                "image_url": {"url": f"data:{mime};base64,{b64}"},
-                "_meta": {"path": str(p)},
-            })
 
-        if not images:
-            return text
-        return images + [{"type": "text", "text": text}]
+            size_ok = self._file_size_ok(p, limits.max_input_image_bytes)
+            if size_ok is False:
+                size_mb = limits.max_input_image_bytes // (1024 * 1024)
+                notes.append(f"[Skipped image: file too large ({p.name}, limit {size_mb} MB)]")
+                continue
+            try:
+                raw = p.read_bytes()
+            except OSError:
+                notes.append(f"[Skipped image: unable to read ({p.name or path})]")
+                continue
+            img_mime = detect_image_mime(raw[:32]) or mimetypes.guess_type(path)[0]
+            if not img_mime or not img_mime.startswith("image/"):
+                notes.append(f"[Skipped image: unsupported or invalid image format ({p.name})]")
+                continue
+            blocks.append(self._encode_image_block(raw, img_mime, p))
+
+        # Process non-image media (audio, video, unknown)
+        audio_count = 0
+        video_count = 0
+        for path in non_image_media:
+            p = Path(path)
+            if not p.is_file():
+                continue
+            guessed_mime = mimetypes.guess_type(path)[0] or ""
+            is_audio = guessed_mime.startswith("audio/")
+            is_video = guessed_mime.startswith("video/")
+
+            # Pre-check file size via stat to avoid reading oversized files into memory.
+            # Determine the relevant byte limit based on detected media type.
+            _size_limit = 0
+            if is_audio or is_video:
+                _size_limit = limits.max_input_audio_bytes if is_audio else limits.max_input_video_bytes
+            _stat_size_ok = self._file_size_ok(p, _size_limit) if _size_limit else None
+            if _stat_size_ok is False:
+                size_mb = _size_limit // (1024 * 1024)
+                label = "audio" if is_audio else "video"
+                notes.append(f"[Skipped {label}: file too large ({p.name}, limit {size_mb} MB)]")
+                continue
+
+            try:
+                raw = p.read_bytes()
+            except OSError:
+                notes.append(f"[Skipped file: unable to read ({p.name or path})]")
+                continue
+
+            # Audio detection: by magic bytes or by filename
+            # Always pass filename so fallback can match when magic bytes fail
+            audio_mime = detect_audio_mime(raw[:32], filename=path)
+            if audio_mime or is_audio:
+                if supports_audio is True and audio_mime_compat(audio_mime):
+                    audio_count += 1
+                    if audio_count > limits.max_input_audios:
+                        if audio_count == limits.max_input_audios + 1:
+                            notes.append(
+                                f"[Skipped audio: only {limits.max_input_audios} audio file(s) allowed]"
+                            )
+                        continue
+                    if len(raw) > limits.max_input_audio_bytes:
+                        size_mb = limits.max_input_audio_bytes // (1024 * 1024)
+                        notes.append(f"[Skipped audio: file too large ({p.name}, limit {size_mb} MB)]")
+                        continue
+                    b64 = base64.b64encode(raw).decode()
+                    blocks.append({
+                        "type": "input_audio",
+                        "input_audio": {"data": b64, "format": audio_format_for_api(audio_mime)},
+                        "_meta": {"path": str(p)},
+                    })
+                else:
+                    blocks.append({"type": "text", "text": f"[audio: {p}]"})
+                continue
+
+            # Video detection (already classified above)
+            if is_video:
+                if supports_video is True and video_mime_compat(guessed_mime):
+                    video_count += 1
+                    if video_count > limits.max_input_videos:
+                        if video_count == limits.max_input_videos + 1:
+                            notes.append(
+                                f"[Skipped video: only {limits.max_input_videos} video file(s) allowed]"
+                            )
+                        continue
+                    if len(raw) > limits.max_input_video_bytes:
+                        size_mb = limits.max_input_video_bytes // (1024 * 1024)
+                        notes.append(f"[Skipped video: file too large ({p.name}, limit {size_mb} MB)]")
+                        continue
+                    b64 = base64.b64encode(raw).decode()
+                    blocks.append({
+                        "type": "video_url",
+                        "video_url": {"url": f"data:{guessed_mime};base64,{b64}"},
+                        "_meta": {"path": str(p)},
+                    })
+                else:
+                    blocks.append({"type": "text", "text": f"[video: {p}]"})
+                continue
+
+            # Unknown files are silently ignored (preserves pre-multimodal behaviour)
+            continue
+
+        note_text = "\n".join(notes).strip()
+        text_block = text if not note_text else (f"{note_text}\n\n{text}" if text else note_text)
+
+        if not blocks:
+            return text_block
+        return blocks + [{"type": "text", "text": text_block}]
 

nanobot/agent/loop.py

@@ -190,6 +190,10 @@ class AgentLoop:
         model_preset: str | None = None,
         preset_snapshot_loader: preset_helpers.PresetSnapshotLoader | None = None,
         runtime_model_publisher: Callable[[str, str | None], None] | None = None,
+        input_limits: Any = None,
+        supports_vision: bool | None = None,
+        supports_audio: bool | None = None,
+        supports_video: bool | None = None,
     ):
         from nanobot.config.schema import ToolsConfig
 
@@ -227,6 +231,10 @@ class AgentLoop:
         self.tools_config = _tc
         self.web_config = _tc.web
         self.exec_config = _tc.exec
+        self.input_limits = input_limits or _tc.input_limits
+        self._supports_vision = supports_vision
+        self._supports_audio = supports_audio
+        self._supports_video = supports_video
         self._image_generation_provider_configs = dict(image_generation_provider_configs or {})
         if (
             image_generation_provider_config is not None
@@ -240,7 +248,7 @@ class AgentLoop:
         self._pending_turn_latency_ms: dict[str, int] = {}
         self._extra_hooks: list[AgentHook] = hooks or []
 
-        self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills)
+        self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills, input_limits=self.input_limits)
         self.sessions = session_manager or SessionManager(workspace)
         self._webui_turns = WebuiTurnCoordinator(
             bus=self.bus,
@@ -366,6 +374,10 @@ class AgentLoop:
             model_preset=defaults.model_preset,
             provider_snapshot_loader=provider_snapshot_loader,
             preset_snapshot_loader=preset_snapshot_loader,
+            input_limits=config.tools.input_limits,
+            supports_vision=defaults.supports_vision(defaults.model),
+            supports_audio=defaults.supports_audio(defaults.model),
+            supports_video=defaults.supports_video(defaults.model),
             **extra,
         )
 
@@ -594,6 +606,9 @@ class AgentLoop:
             sender_id=msg.sender_id,
             session_summary=pending_summary,
             session_metadata=session.metadata,
+            supports_vision=self._supports_vision,
+            supports_audio=self._supports_audio,
+            supports_video=self._supports_video,
         )
 
     async def _dispatch_command_inline(
@@ -1059,6 +1074,9 @@ class AgentLoop:
             sender_id=msg.sender_id,
             session_summary=pending,
             session_metadata=session.metadata,
+            supports_vision=self._supports_vision,
+            supports_audio=self._supports_audio,
+            supports_video=self._supports_video,
         )
         t_wall = time.time()
         final_content, _, all_msgs, stop_reason, _ = await self._run_agent_loop(
@@ -1404,6 +1422,10 @@ class AgentLoop:
                 filtered.append({"type": "text", "text": image_placeholder_text(path)})
                 continue
 
+            if block.get("type") in ("input_audio", "video_url"):
+                filtered.append(LLMProvider._media_placeholder(block["type"], block))
+                continue
+
             if block.get("type") == "text" and isinstance(block.get("text"), str):
                 text = block["text"]
                 if should_truncate_text and len(text) > self.max_tool_result_chars:

nanobot/agent/tools/shell.py

@@ -266,7 +266,6 @@ class ExecTool(Tool):
             # the raw command string to COMSPEC without re-quoting.
             return await asyncio.create_subprocess_shell(
                 command,
-                stdin=asyncio.subprocess.DEVNULL,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
                 cwd=cwd,
@@ -275,7 +274,6 @@ class ExecTool(Tool):
         bash = shutil.which("bash") or "/bin/bash"
         return await asyncio.create_subprocess_exec(
             bash, "-l", "-c", command,
-            stdin=asyncio.subprocess.DEVNULL,
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
             cwd=cwd,

nanobot/channels/feishu.py

@@ -172,19 +172,22 @@ def _extract_element_content(element: dict) -> list[str]:
     return parts
 
 
-def _extract_post_content(content_json: dict) -> tuple[str, list[str]]:
-    """Extract text and image keys from Feishu post (rich text) message.
+def _extract_post_content(content_json: dict) -> tuple[str, list[str], list[dict]]:
+    """Extract text and media info from Feishu post (rich text) message.
 
     Handles three payload shapes:
     - Direct:    {"title": "...", "content": [[...]]}
     - Localized: {"zh_cn": {"title": "...", "content": [...]}}
     - Wrapped:   {"post": {"zh_cn": {"title": "...", "content": [...]}}}
+
+    Returns (text, image_keys, media_items) where media_items is a list of
+    {"tag": "media", "file_key": "..."} dicts for video/file attachments.
     """
 
-    def _parse_block(block: dict) -> tuple[str | None, list[str]]:
+    def _parse_block(block: dict) -> tuple[str | None, list[str], list[dict]]:
         if not isinstance(block, dict) or not isinstance(block.get("content"), list):
-            return None, []
-        texts, images = [], []
+            return None, [], []
+        texts, images, medias = [], [], []
         if title := block.get("title"):
             texts.append(title)
         for row in block["content"]:
@@ -204,43 +207,36 @@ def _extract_post_content(content_json: dict) -> tuple[str, list[str]]:
                     texts.append(f"\n```{lang}\n{code_text}\n```\n")
                 elif tag == "img" and (key := el.get("image_key")):
                     images.append(key)
-        return (" ".join(texts).strip() or None), images
+                elif tag == "media" and el.get("file_key"):
+                    medias.append({"tag": "media", "file_key": el["file_key"]})
+        return (" ".join(texts).strip() or None), images, medias
 
     # Unwrap optional {"post": ...} envelope
     root = content_json
     if isinstance(root, dict) and isinstance(root.get("post"), dict):
         root = root["post"]
     if not isinstance(root, dict):
-        return "", []
+        return "", [], []
 
     # Direct format
     if "content" in root:
-        text, imgs = _parse_block(root)
-        if text or imgs:
-            return text or "", imgs
+        text, imgs, medias = _parse_block(root)
+        if text or imgs or medias:
+            return text or "", imgs, medias
 
     # Localized: prefer known locales, then fall back to any dict child
     for key in ("zh_cn", "en_us", "ja_jp"):
         if key in root:
-            text, imgs = _parse_block(root[key])
-            if text or imgs:
-                return text or "", imgs
+            text, imgs, medias = _parse_block(root[key])
+            if text or imgs or medias:
+                return text or "", imgs, medias
     for val in root.values():
         if isinstance(val, dict):
-            text, imgs = _parse_block(val)
-            if text or imgs:
-                return text or "", imgs
+            text, imgs, medias = _parse_block(val)
+            if text or imgs or medias:
+                return text or "", imgs, medias
 
-    return "", []
-
-
-def _extract_post_text(content_json: dict) -> str:
-    """Extract plain text from Feishu post (rich text) message content.
-
-    Legacy wrapper for _extract_post_content, returns only text.
-    """
-    text, _ = _extract_post_content(content_json)
-    return text
+    return "", [], []
 
 
 class FeishuConfig(Base):
@@ -1156,7 +1152,7 @@ class FeishuChannel(BaseChannel):
             if msg_type == "text":
                 text = content_json.get("text", "").strip()
             elif msg_type == "post":
-                text, _ = _extract_post_content(content_json)
+                text, _, _ = _extract_post_content(content_json)
                 text = text.strip()
             else:
                 text = ""
@@ -1751,7 +1747,7 @@ class FeishuChannel(BaseChannel):
                     content_parts.append(text)
 
             elif msg_type == "post":
-                text, image_keys = _extract_post_content(content_json)
+                text, image_keys, media_items = _extract_post_content(content_json)
                 if text:
                     content_parts.append(text)
                 # Download images embedded in post
@@ -1762,6 +1758,14 @@ class FeishuChannel(BaseChannel):
                     if file_path:
                         media_paths.append(file_path)
                     content_parts.append(content_text)
+                # Download media (video/file) embedded in post
+                for media_item in media_items:
+                    file_path, content_text = await self._download_and_save_media(
+                        "media", media_item, message_id
+                    )
+                    if file_path:
+                        media_paths.append(file_path)
+                    content_parts.append(content_text)
 
             elif msg_type in ("image", "audio", "file", "media"):
                 file_path, content_text = await self._download_and_save_media(

nanobot/channels/manager.py

@@ -70,40 +70,28 @@ class ChannelManager:
 
     def _init_channels(self) -> None:
         """Initialize channels discovered via pkgutil scan + entry_points plugins."""
-        from nanobot.channels.registry import discover_channel_names, discover_enabled
+        from nanobot.channels.registry import discover_all
 
         transcription_provider = self.config.channels.transcription_provider
         transcription_key = self._resolve_transcription_key(transcription_provider)
         transcription_base = self._resolve_transcription_base(transcription_provider)
         transcription_language = self.config.channels.transcription_language
 
-        # Collect enabled module names first, then only import those.
-        # Channel configs live in ChannelsConfig's extra fields (via
-        # extra="allow"), so we enumerate candidates from pkgutil scan
-        # (cheap, no imports) and any plugin keys in __pydantic_extra__.
-        names = discover_channel_names()
-        candidate_names = set(names)
-        extra = getattr(self.config.channels, "__pydantic_extra__", None) or {}
-        candidate_names.update(extra.keys())
-
-        enabled_names: set[str] = set()
-        for name in candidate_names:
+        for name, cls in discover_all().items():
             section = getattr(self.config.channels, name, None)
             if section is None:
                 continue
-            if (
+            enabled = (
                 section.get("enabled", False)
                 if isinstance(section, dict)
                 else getattr(section, "enabled", False)
-            ):
-                enabled_names.add(name)
-
-        for name, cls in discover_enabled(enabled_names, _names=names).items():
-            section = getattr(self.config.channels, name, None)
-            if section is None:
+            )
+            if not enabled:
                 continue
             try:
                 kwargs: dict[str, Any] = {}
+                # Only the WebSocket channel currently hosts the embedded webui
+                # surface; other channels stay oblivious to these knobs.
                 if cls.name == "websocket":
                     if self._session_manager is not None:
                         kwargs["session_manager"] = self._session_manager

nanobot/channels/registry.py

@@ -1,4 +1,5 @@
 """Auto-discovery for built-in channel modules and external plugins."""
+
 from __future__ import annotations
 
 import importlib
@@ -36,14 +37,12 @@ def load_channel_class(module_name: str) -> type[BaseChannel]:
     raise ImportError(f"No BaseChannel subclass in nanobot.channels.{module_name}")
 
 
-def discover_plugins(enabled_names: set[str] | None = None) -> dict[str, type[BaseChannel]]:
+def discover_plugins() -> dict[str, type[BaseChannel]]:
     """Discover external channel plugins registered via entry_points."""
     from importlib.metadata import entry_points
 
     plugins: dict[str, type[BaseChannel]] = {}
     for ep in entry_points(group="nanobot.channels"):
-        if enabled_names is not None and ep.name not in enabled_names:
-            continue
         try:
             cls = ep.load()
             plugins[ep.name] = cls
@@ -52,44 +51,21 @@ def discover_plugins(enabled_names: set[str] | None = None) -> dict[str, type[Ba
     return plugins
 
 
-def discover_enabled(
-    enabled_names: set[str],
-    *,
-    _names: list[str] | None = None,
-    _include_all_external: bool = False,
-) -> dict[str, type[BaseChannel]]:
-    """Return channels whose module names are in *enabled_names*.
-
-    Uses cheap ``pkgutil.iter_modules`` to list names, then imports only
-    those that match — skipping the heavy third-party SDK imports of
-    unneeded channels.
-    """
-    names = _names if _names is not None else discover_channel_names()
-    result: dict[str, type[BaseChannel]] = {}
-    for modname in names:
-        if modname not in enabled_names:
-            continue
-        try:
-            result[modname] = load_channel_class(modname)
-        except ImportError as e:
-            logger.debug("Skipping built-in channel '{}': {}", modname, e)
-
-    external = discover_plugins(None if _include_all_external else enabled_names)
-    shadowed = set(external) & set(result)
-    if shadowed:
-        logger.warning("Plugin(s) shadowed by built-in channels (ignored): {}", shadowed)
-    if _include_all_external:
-        result.update({k: v for k, v in external.items() if k not in shadowed})
-    else:
-        result.update({k: v for k, v in external.items() if k not in shadowed and k in enabled_names})
-
-    return result
-
-
 def discover_all() -> dict[str, type[BaseChannel]]:
     """Return all channels: built-in (pkgutil) merged with external (entry_points).
 
     Built-in channels take priority — an external plugin cannot shadow a built-in name.
     """
-    names = discover_channel_names()
-    return discover_enabled(set(names), _names=names, _include_all_external=True)
+    builtin: dict[str, type[BaseChannel]] = {}
+    for modname in discover_channel_names():
+        try:
+            builtin[modname] = load_channel_class(modname)
+        except ImportError as e:
+            logger.debug("Skipping built-in channel '{}': {}", modname, e)
+
+    external = discover_plugins()
+    shadowed = set(external) & set(builtin)
+    if shadowed:
+        logger.warning("Plugin(s) shadowed by built-in channels (ignored): {}", shadowed)
+
+    return {**external, **builtin}

nanobot/cli/commands.py

@@ -1,14 +1,12 @@
 """CLI commands for nanobot."""
 
 import asyncio
-import json
 import os
 import select
 import signal
 import sys
 from collections.abc import Callable
 from contextlib import nullcontext, suppress
-from inspect import signature
 from pathlib import Path
 from typing import Any
 
@@ -1529,106 +1527,6 @@ def status():
                 console.print(f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}")
 
 
-# ============================================================================
-# Config Commands
-# ============================================================================
-
-config_app = typer.Typer(help="Manage configuration")
-app.add_typer(config_app, name="config")
-
-
-@config_app.command("set")
-def config_set(
-    path: str = typer.Argument(..., help="Dot path, e.g. agents.defaults.model"),
-    value: str = typer.Argument(..., help="Value. Use null/true/false or JSON for structured values."),
-    config_path: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
-):
-    """Set one config value by dot path."""
-    from pydantic import ValidationError
-
-    from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
-    from nanobot.config.schema import Config
-
-    resolved_path = Path(config_path).expanduser().resolve() if config_path else get_config_path()
-    if config_path:
-        set_config_path(resolved_path)
-
-    config = load_config(resolved_path)
-    parsed = _parse_config_cli_value(value)
-    try:
-        _set_config_cli_value(config, path, parsed)
-        validated = Config.model_validate(config.model_dump(mode="json", by_alias=True))
-    except (AttributeError, KeyError, TypeError, ValueError, ValidationError) as exc:
-        console.print(f"[red]Could not set config value:[/red] {exc}")
-        raise typer.Exit(1)
-
-    save_config(validated, resolved_path)
-    console.print(f"[green]✓[/green] Set [cyan]{path}[/cyan] = [bold]{value}[/bold]")
-    console.print(f"[dim]Config: {resolved_path}[/dim]")
-    if path in {"agents.defaults.provider", "agents.defaults.model"} and validated.agents.defaults.model_preset:
-        console.print(
-            "[yellow]! agents.defaults.model_preset is set and may override this. "
-            "Clear it with: nanobot config set agents.defaults.model_preset null[/yellow]"
-        )
-
-
-def _parse_config_cli_value(raw: str) -> Any:
-    lowered = raw.strip().lower()
-    if lowered == "null":
-        return None
-    if lowered == "true":
-        return True
-    if lowered == "false":
-        return False
-    with suppress(Exception):
-        return json.loads(raw)
-    return raw
-
-
-def _resolve_config_field(obj: Any, key: str) -> str:
-    from pydantic import BaseModel
-    from pydantic.alias_generators import to_camel, to_snake
-
-    if not isinstance(obj, BaseModel):
-        return key
-    fields = type(obj).model_fields
-    if key in fields:
-        return key
-    normalized = to_snake(key.replace("-", "_"))
-    if normalized in fields:
-        return normalized
-    for name, field in fields.items():
-        aliases = {
-            to_camel(name),
-            str(field.alias) if field.alias else "",
-            str(field.serialization_alias) if field.serialization_alias else "",
-        }
-        if key in aliases:
-            return name
-    raise AttributeError(f"Unknown config path segment {key!r}")
-
-
-def _set_config_cli_value(config: Any, path: str, value: Any) -> None:
-    parts = [part for part in path.split(".") if part]
-    if not parts:
-        raise ValueError("Config path cannot be empty.")
-
-    current = config
-    for raw_part in parts[:-1]:
-        if isinstance(current, dict):
-            current = current.setdefault(raw_part, {})
-            continue
-        part = _resolve_config_field(current, raw_part)
-        current = getattr(current, part)
-
-    leaf = parts[-1]
-    if isinstance(current, dict):
-        current[leaf] = value
-        return
-    leaf = _resolve_config_field(current, leaf)
-    setattr(current, leaf, value)
-
-
 # ============================================================================
 # OAuth Login
 # ============================================================================
@@ -1643,7 +1541,6 @@ _LOGOUT_HANDLERS: dict[str, Callable[[], None]] = {}
 _PROVIDER_DISPLAY: dict[str, str] = {
     "openai_codex": "OpenAI Codex",
     "github_copilot": "GitHub Copilot",
-    "xai_oauth": "xAI Grok OAuth",
 }
 
 
@@ -1679,9 +1576,7 @@ def _resolve_oauth_provider(provider: str):
 
 @provider_app.command("login")
 def provider_login(
-    provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot', 'xai-oauth')"),
-    no_browser: bool = typer.Option(False, "--no-browser", help="Print the auth URL instead of opening a browser when supported."),
-    manual_paste: bool = typer.Option(False, "--manual-paste", help="Prompt for a callback URL or fallback code when supported."),
+    provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot')"),
 ):
     """Authenticate with an OAuth provider."""
     spec = _resolve_oauth_provider(provider)
@@ -1692,18 +1587,12 @@ def provider_login(
         raise typer.Exit(1)
 
     console.print(f"{__logo__} OAuth Login - {spec.label}\n")
-    params = signature(handler).parameters
-    kwargs: dict[str, bool] = {}
-    if "no_browser" in params:
-        kwargs["no_browser"] = no_browser
-    if "manual_paste" in params:
-        kwargs["manual_paste"] = manual_paste
-    handler(**kwargs)
+    handler()
 
 
 @provider_app.command("logout")
 def provider_logout(
-    provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot', 'xai-oauth')"),
+    provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot')"),
 ):
     """Log out from an OAuth provider."""
     spec = _resolve_oauth_provider(provider)
@@ -1767,24 +1656,6 @@ def _logout_github_copilot() -> None:
     _delete_oauth_files(storage.get_token_path(), _PROVIDER_DISPLAY["github_copilot"])
 
 
-@_register_logout("xai_oauth")
-def _logout_xai_oauth() -> None:
-    """Clear local OAuth credentials for xAI Grok OAuth."""
-    try:
-        from nanobot.providers.xai_oauth_provider import delete_xai_oauth_credentials
-    except ImportError:
-        console.print("[red]xAI Grok OAuth provider unavailable.[/red]")
-        raise typer.Exit(1)
-
-    removed_paths = delete_xai_oauth_credentials()
-    if not removed_paths:
-        console.print(f"[yellow]! No local OAuth credentials found for {_PROVIDER_DISPLAY['xai_oauth']}[/yellow]")
-        return
-    console.print(f"[green]✓ Logged out from {_PROVIDER_DISPLAY['xai_oauth']}[/green]")
-    for path in removed_paths:
-        console.print(f"[dim]Removed: {path}[/dim]")
-
-
 def _delete_oauth_files(token_path: Path, provider_label: str) -> None:
     """Delete OAuth token and lock files, reporting the result."""
     removed_paths: list[Path] = []
@@ -1828,36 +1699,5 @@ def _login_github_copilot() -> None:
         raise typer.Exit(1)
 
 
-@_register_login("xai_oauth")
-def _login_xai_oauth(
-    *,
-    no_browser: bool = False,
-    manual_paste: bool = False,
-) -> None:
-    try:
-        from nanobot.providers.xai_oauth_provider import login_xai_oauth_interactive
-        from nanobot.providers.xai_oauth_provider import DEFAULT_XAI_MODEL
-
-        console.print("[cyan]Starting xAI Grok OAuth login...[/cyan]\n")
-        credential = login_xai_oauth_interactive(
-            print_fn=lambda s: console.print(s),
-            prompt_fn=lambda s: typer.prompt(s),
-            open_browser=not no_browser,
-            manual_paste=manual_paste,
-        )
-        account = credential.account_id or "xAI"
-        storage = "OS keychain" if credential.storage == "keyring" else "private file"
-        console.print(f"[green]✓ Authenticated with xAI Grok OAuth[/green]  [dim]{account} · {storage}[/dim]")
-        console.print("[dim]To use it for chat:[/dim]")
-        console.print("[dim]  nanobot config set agents.defaults.model_preset null[/dim]")
-        console.print("[dim]  nanobot config set agents.defaults.provider xai-oauth[/dim]")
-        console.print(f"[dim]  nanobot config set agents.defaults.model {DEFAULT_XAI_MODEL}[/dim]")
-        console.print("[dim]Hosted X Search is enabled by default for xAI OAuth.[/dim]")
-        console.print("[dim]To disable it: nanobot config set providers.xai_oauth.x_search.enable false[/dim]")
-    except Exception as e:
-        console.print(f"[red]Authentication error: {e}[/red]")
-        raise typer.Exit(1)
-
-
 if __name__ == "__main__":
     app()

nanobot/config/schema.py

@@ -155,8 +155,35 @@ class AgentDefaults(Base):
         validation_alias=AliasChoices("consolidationRatio"),
         serialization_alias="consolidationRatio",
     )  # Consolidation target ratio (0.5 = 50% of budget retained after compression)
+    vision_models: list[str] = Field(default_factory=list)  # Models that support image input
+    audio_models: list[str] = Field(default_factory=list)  # Models that support native audio input
+    video_models: list[str] = Field(default_factory=list)  # Models that support native video input
     dream: DreamConfig = Field(default_factory=DreamConfig)
 
+    @staticmethod
+    def _bare_model(model: str) -> str:
+        """Strip provider prefix, e.g. 'openai/gpt-4o' -> 'gpt-4o'."""
+        return model.split("/", 1)[-1].lower() if "/" in model else model.lower()
+
+    def _supports_capability(self, model: str, patterns: list[str]) -> bool | None:
+        """Check if model matches any pattern. Returns None if patterns is empty."""
+        if not patterns:
+            return None
+        bare = self._bare_model(model)
+        return any(p.lower() in bare for p in patterns)
+
+    def supports_vision(self, model: str) -> bool | None:
+        """Check if model supports vision. None if unconfigured."""
+        return self._supports_capability(model, self.vision_models)
+
+    def supports_audio(self, model: str) -> bool | None:
+        """Check if model supports native audio. None if unconfigured."""
+        return self._supports_capability(model, self.audio_models)
+
+    def supports_video(self, model: str) -> bool | None:
+        """Check if model supports native video. None if unconfigured."""
+        return self._supports_capability(model, self.video_models)
+
 
 class AgentsConfig(Base):
     """Agent configuration."""
@@ -180,28 +207,6 @@ class BedrockProviderConfig(ProviderConfig):
     profile: str | None = None  # Optional AWS shared config profile
 
 
-class XaiOAuthXSearchConfig(Base):
-    """xAI hosted X Search configuration."""
-
-    enable: bool = True
-    allowed_x_handles: list[str] | None = None
-    excluded_x_handles: list[str] | None = None
-    from_date: str | None = None
-    to_date: str | None = None
-    enable_image_understanding: bool = False
-    enable_video_understanding: bool = False
-
-
-class XaiOAuthProviderConfig(ProviderConfig):
-    """xAI OAuth provider configuration."""
-
-    x_search: XaiOAuthXSearchConfig = Field(default_factory=XaiOAuthXSearchConfig)
-
-
-def _is_default_xai_oauth_config(value: Any) -> bool:
-    return isinstance(value, XaiOAuthProviderConfig) and value == XaiOAuthProviderConfig()
-
-
 class ProvidersConfig(Base):
     """Configuration for LLM providers."""
 
@@ -239,10 +244,6 @@ class ProvidersConfig(Base):
     byteplus_coding_plan: ProviderConfig = Field(default_factory=ProviderConfig)  # BytePlus Coding Plan
     openai_codex: ProviderConfig = Field(default_factory=ProviderConfig, exclude=True)  # OpenAI Codex (OAuth)
     github_copilot: ProviderConfig = Field(default_factory=ProviderConfig, exclude=True)  # Github Copilot (OAuth)
-    xai_oauth: XaiOAuthProviderConfig = Field(
-        default_factory=XaiOAuthProviderConfig,
-        exclude_if=_is_default_xai_oauth_config,
-    )  # xAI Grok OAuth
     qianfan: ProviderConfig = Field(default_factory=ProviderConfig)  # Qianfan (百度千帆)
     nvidia: ProviderConfig = Field(default_factory=ProviderConfig)  # NVIDIA NIM (nvapi- keys)
 
@@ -284,6 +285,17 @@ class MCPServerConfig(Base):
     enabled_tools: list[str] = Field(default_factory=lambda: ["*"])  # Only register these tools; accepts raw MCP names or wrapped mcp_<server>_<tool> names; ["*"] = all tools; [] = no tools
 
 
+class InputLimitsConfig(Base):
+    """Limits for user-provided multimodal inputs."""
+
+    max_input_images: int = 3
+    max_input_image_bytes: int = 10 * 1024 * 1024  # 10 MB
+    max_input_audios: int = 1
+    max_input_audio_bytes: int = 10 * 1024 * 1024  # 10 MB
+    max_input_videos: int = 1
+    max_input_video_bytes: int = 20 * 1024 * 1024  # 20 MB
+
+
 def _lazy_default(module_path: str, class_name: str) -> Any:
     """Deferred import helper for ToolsConfig default factories."""
     import importlib
@@ -305,6 +317,7 @@ class ToolsConfig(Base):
     image_generation: ImageGenerationToolConfig = Field(
         default_factory=lambda: _lazy_default("nanobot.agent.tools.image_generation", "ImageGenerationToolConfig"),
     )
+    input_limits: InputLimitsConfig = Field(default_factory=InputLimitsConfig)
     restrict_to_workspace: bool = False  # restrict all tool access to workspace directory
     mcp_servers: dict[str, MCPServerConfig] = Field(default_factory=dict)
     ssrf_whitelist: list[str] = Field(default_factory=list)  # CIDR ranges to exempt from SSRF blocking (e.g. ["100.64.0.0/10"] for Tailscale)

nanobot/cron/__init__.py

@@ -1,18 +1,6 @@
 """Cron service for scheduled agent tasks."""
 
+from nanobot.cron.service import CronService
 from nanobot.cron.types import CronJob, CronSchedule
 
 __all__ = ["CronService", "CronJob", "CronSchedule"]
-
-_LAZY = {"CronService": ".service"}
-
-
-def __getattr__(name: str):
-    module_path = _LAZY.get(name)
-    if module_path is None:
-        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-    from importlib import import_module
-    mod = import_module(module_path, __name__)
-    val = getattr(mod, name)
-    globals()[name] = val
-    return val

nanobot/providers/__init__.py

@@ -14,7 +14,6 @@ __all__ = [
     "OpenAICompatProvider",
     "OpenAICodexProvider",
     "GitHubCopilotProvider",
-    "XaiOAuthProvider",
     "AzureOpenAIProvider",
     "BedrockProvider",
 ]
@@ -24,23 +23,10 @@ _LAZY_IMPORTS = {
     "OpenAICompatProvider": ".openai_compat_provider",
     "OpenAICodexProvider": ".openai_codex_provider",
     "GitHubCopilotProvider": ".github_copilot_provider",
-    "XaiOAuthProvider": ".xai_oauth_provider",
     "AzureOpenAIProvider": ".azure_openai_provider",
     "BedrockProvider": ".bedrock_provider",
 }
 
-_LAZY_SUBMODULES = {
-    "anthropic_provider": ".anthropic_provider",
-    "openai_compat_provider": ".openai_compat_provider",
-    "openai_codex_provider": ".openai_codex_provider",
-    "github_copilot_provider": ".github_copilot_provider",
-    "xai_oauth_provider": ".xai_oauth_provider",
-    "azure_openai_provider": ".azure_openai_provider",
-    "bedrock_provider": ".bedrock_provider",
-    "factory": ".factory",
-    "registry": ".registry",
-}
-
 if TYPE_CHECKING:
     from nanobot.providers.anthropic_provider import AnthropicProvider
     from nanobot.providers.azure_openai_provider import AzureOpenAIProvider
@@ -48,18 +34,12 @@ if TYPE_CHECKING:
     from nanobot.providers.github_copilot_provider import GitHubCopilotProvider
     from nanobot.providers.openai_compat_provider import OpenAICompatProvider
     from nanobot.providers.openai_codex_provider import OpenAICodexProvider
-    from nanobot.providers.xai_oauth_provider import XaiOAuthProvider
 
 
 def __getattr__(name: str):
     """Lazily expose provider implementations without importing all backends up front."""
     module_name = _LAZY_IMPORTS.get(name)
-    if module_name is not None:
-        module = import_module(module_name, __name__)
-        return getattr(module, name)
-    module_name = _LAZY_SUBMODULES.get(name)
-    if module_name is not None:
-        module = import_module(module_name, __name__)
-        globals()[name] = module
-        return module
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    if module_name is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    module = import_module(module_name, __name__)
+    return getattr(module, name)

nanobot/providers/anthropic_provider.py

@@ -212,7 +212,7 @@ class AnthropicProvider(LLMProvider):
 
     @staticmethod
     def _convert_user_content(content: Any) -> Any:
-        """Convert user message content, translating image_url blocks."""
+        """Convert user message content, translating image_url and input_audio blocks."""
         if isinstance(content, str) or content is None:
             return content or "(empty)"
         if not isinstance(content, list):
@@ -228,6 +228,14 @@ class AnthropicProvider(LLMProvider):
                 if converted:
                     result.append(converted)
                 continue
+            if item.get("type") == "input_audio":
+                # Anthropic doesn't support native audio → text placeholder
+                result.append(LLMProvider._media_placeholder("input_audio", item))
+                continue
+            if item.get("type") == "video_url":
+                # Anthropic doesn't support native video → text placeholder
+                result.append(LLMProvider._media_placeholder("video_url", item))
+                continue
             result.append(item)
         return result or "(empty)"
 

nanobot/providers/base.py

@@ -13,8 +13,6 @@ from typing import Any
 
 from loguru import logger
 
-from nanobot.utils.helpers import image_placeholder_text
-
 
 @dataclass
 class ToolCallRequest:
@@ -439,9 +437,23 @@ class LLMProvider(ABC):
 
         return merged
 
+    _MEDIA_LABEL_MAP = {"image_url": "image", "input_audio": "audio", "video_url": "video"}
+    _STRIP_MEDIA_TYPES = frozenset({"image_url", "input_audio", "video_url"})
+
     @staticmethod
-    def _strip_image_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
-        """Replace image_url blocks with text placeholder. Returns None if no images found."""
+    def _media_placeholder(btype: str, block: dict[str, Any]) -> dict[str, str]:
+        """Build a text placeholder for a media block."""
+        path = (block.get("_meta") or {}).get("path", "")
+        label = LLMProvider._MEDIA_LABEL_MAP.get(btype, "media")
+        text = f"[{label}: {path}]" if path else f"[{label}]"
+        return {"type": "text", "text": text}
+
+    @staticmethod
+    def _strip_media_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
+        """Replace image_url, input_audio, and video_url blocks with text placeholders.
+
+        Returns None if no media blocks were found (no changes needed).
+        """
         found = False
         result = []
         for msg in messages:
@@ -449,10 +461,8 @@ class LLMProvider(ABC):
             if isinstance(content, list):
                 new_content = []
                 for b in content:
-                    if isinstance(b, dict) and b.get("type") == "image_url":
-                        path = (b.get("_meta") or {}).get("path", "")
-                        placeholder = image_placeholder_text(path, empty="[image omitted]")
-                        new_content.append({"type": "text", "text": placeholder})
+                    if isinstance(b, dict) and b.get("type") in LLMProvider._STRIP_MEDIA_TYPES:
+                        new_content.append(LLMProvider._media_placeholder(b["type"], b))
                         found = True
                     else:
                         new_content.append(b)
@@ -462,8 +472,13 @@ class LLMProvider(ABC):
         return result if found else None
 
     @staticmethod
-    def _strip_image_content_inplace(messages: list[dict[str, Any]]) -> bool:
-        """Replace image_url blocks with text placeholder *in-place*.
+    def _strip_image_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
+        """Replace image_url blocks with text placeholder. Returns None if no images found."""
+        return LLMProvider._strip_media_content(messages)
+
+    @staticmethod
+    def _strip_media_content_inplace(messages: list[dict[str, Any]]) -> bool:
+        """Replace media blocks with text placeholder *in-place*.
 
         Mutates the content lists of the original message dicts so that
         callers holding references to those dicts also see the stripped
@@ -474,13 +489,16 @@ class LLMProvider(ABC):
             content = msg.get("content")
             if isinstance(content, list):
                 for i, b in enumerate(content):
-                    if isinstance(b, dict) and b.get("type") == "image_url":
-                        path = (b.get("_meta") or {}).get("path", "")
-                        placeholder = image_placeholder_text(path, empty="[image omitted]")
-                        content[i] = {"type": "text", "text": placeholder}
+                    if isinstance(b, dict) and b.get("type") in LLMProvider._STRIP_MEDIA_TYPES:
+                        content[i] = LLMProvider._media_placeholder(b["type"], b)
                         found = True
         return found
 
+    @staticmethod
+    def _strip_image_content_inplace(messages: list[dict[str, Any]]) -> bool:
+        """Replace image_url blocks with text placeholder *in-place*."""
+        return LLMProvider._strip_media_content_inplace(messages)
+
     async def _safe_chat(self, **kwargs: Any) -> LLMResponse:
         """Call chat() and convert unexpected exceptions to error responses."""
         try:
@@ -738,18 +756,18 @@ class LLMProvider(ABC):
                 identical_error_count = 1 if error_key else 0
 
             if not self._is_transient_response(response):
-                stripped = self._strip_image_content(original_messages)
+                stripped = self._strip_media_content(original_messages)
                 if stripped is not None and stripped != kw["messages"]:
                     logger.warning(
-                        "Non-transient LLM error with image content, retrying without images"
+                        "Non-transient LLM error with media content, retrying without media"
                     )
                     retry_kw = dict(kw)
                     retry_kw["messages"] = stripped
                     result = await call(**retry_kw)
-                    # Permanently strip images from the original messages so
+                    # Permanently strip media from the original messages so
                     # subsequent iterations do not repeat the error-retry cycle.
                     if result.finish_reason != "error":
-                        self._strip_image_content_inplace(original_messages)
+                        self._strip_media_content_inplace(original_messages)
                     return result
                 return response
 

nanobot/providers/factory.py

@@ -68,10 +68,6 @@ def _make_provider_core(
         from nanobot.providers.github_copilot_provider import GitHubCopilotProvider
 
         provider = GitHubCopilotProvider(default_model=model)
-    elif backend == "xai_oauth":
-        from nanobot.providers.xai_oauth_provider import XaiOAuthProvider
-
-        provider = XaiOAuthProvider(default_model=model, config=p)
     elif backend == "anthropic":
         from nanobot.providers.anthropic_provider import AnthropicProvider
 

nanobot/providers/github_copilot_provider.py

@@ -207,9 +207,8 @@ class GitHubCopilotProvider(OpenAICompatProvider):
 
     async def _refresh_client_api_key(self) -> str:
         token = await self._get_copilot_access_token()
-        client = await self._ensure_client()
         self.api_key = token
-        client.api_key = token
+        self._client.api_key = token
         return token
 
     async def chat(

nanobot/providers/openai_compat_provider.py

@@ -16,9 +16,20 @@ from ipaddress import ip_address
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlparse
 
+import httpx
 import json_repair
 from loguru import logger
 
+if os.environ.get("LANGFUSE_SECRET_KEY") and importlib.util.find_spec("langfuse"):
+    from langfuse.openai import AsyncOpenAI
+else:
+    if os.environ.get("LANGFUSE_SECRET_KEY"):
+        logger.warning(
+            "LANGFUSE_SECRET_KEY is set but langfuse is not installed; "
+            "install with `pip install langfuse` to enable tracing"
+        )
+    from openai import AsyncOpenAI
+
 from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
 from nanobot.providers.openai_responses import (
     consume_sdk_stream,
@@ -28,15 +39,8 @@ from nanobot.providers.openai_responses import (
 )
 
 if TYPE_CHECKING:
-    from openai import AsyncOpenAI as AsyncOpenAIType
-
     from nanobot.providers.registry import ProviderSpec
 
-# Module-level placeholder — set lazily by _ensure_client on first real
-# use, or replaced by tests via ``patch(...)``.  Kept as a plain name so
-# that ``unittest.mock.patch`` can find and replace it.
-AsyncOpenAI: Any = None
-
 _ALLOWED_MSG_KEYS = frozenset({
     "role", "content", "tool_calls", "tool_call_id", "name",
     "reasoning_content", "extra_content",
@@ -298,75 +302,42 @@ class OpenAICompatProvider(LLMProvider):
 
         effective_base = api_base or (spec.default_api_base if spec else None) or None
         self._effective_base = effective_base
-        self._default_headers = {"x-session-affinity": uuid.uuid4().hex}
+        default_headers = {"x-session-affinity": uuid.uuid4().hex}
         if _uses_openrouter_attribution(spec, effective_base):
-            self._default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
+            default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
         if extra_headers:
-            self._default_headers.update(extra_headers)
-        self._api_key_for_client = api_key or "no-key"
-        self._is_local = _is_local_endpoint(spec, effective_base)
-
-        # Lazy-init: the OpenAI client and its httpx transport are expensive
-        # to create (~700 ms on Windows). Defer until first use.
-        self._client: AsyncOpenAIType | None = None
-        self._client_lock = asyncio.Lock()
-
-        # Responses API circuit breaker: skip after repeated failures,
-        # probe again after _RESPONSES_PROBE_INTERVAL_S seconds.
-        self._responses_failures: dict[str, int] = {}
-        self._responses_tripped_at: dict[str, float] = {}
-
-    def _build_client(self) -> None:
-        """Create the OpenAI client using the current module-level AsyncOpenAI."""
-        import httpx
+            default_headers.update(extra_headers)
 
+        # Local model servers (Ollama, llama.cpp, vLLM) often close idle
+        # HTTP connections before the client-side keepalive expires.  When
+        # two LLM calls happen seconds apart (e.g. heartbeat _decide then
+        # process_direct), the second call may grab a now-dead pooled
+        # connection, causing a transient APIConnectionError on every first
+        # attempt.  Disabling keepalive for local endpoints avoids this by
+        # opening a fresh connection for each request, which is cheap on a
+        # LAN.  Cloud providers benefit from keepalive, so we leave the
+        # default pool settings for them.
         timeout_s = _openai_compat_timeout_s()
         http_client: httpx.AsyncClient | None = None
-        if self._is_local:
-            # Local model servers (Ollama, llama.cpp, vLLM) often close idle
-            # HTTP connections before the client-side keepalive expires. When
-            # two LLM calls happen seconds apart (e.g. heartbeat _decide then
-            # process_direct), the second call may grab a now-dead pooled
-            # connection, causing a transient APIConnectionError on every first
-            # attempt. Disabling keepalive for local endpoints avoids this by
-            # opening a fresh connection for each request, which is cheap on a
-            # LAN. Cloud providers benefit from keepalive, so we leave the
-            # default pool settings for them.
+        if _is_local_endpoint(spec, effective_base):
             http_client = httpx.AsyncClient(
                 limits=httpx.Limits(keepalive_expiry=0),
                 timeout=timeout_s,
             )
+
         self._client = AsyncOpenAI(
-            api_key=self._api_key_for_client,
-            base_url=self._effective_base,
-            default_headers=self._default_headers,
+            api_key=api_key or "no-key",
+            base_url=effective_base,
+            default_headers=default_headers,
             max_retries=0,
             timeout=timeout_s,
             http_client=http_client,
         )
 
-    async def _ensure_client(self):
-        """Return the shared OpenAI client, creating it on first call."""
-        if self._client is not None:
-            return self._client
-        async with self._client_lock:
-            if self._client is not None:
-                return self._client
-            global AsyncOpenAI
-            if AsyncOpenAI is None:
-                if os.environ.get("LANGFUSE_SECRET_KEY") and importlib.util.find_spec("langfuse"):
-                    from langfuse.openai import AsyncOpenAI as _AsyncOpenAI
-                else:
-                    if os.environ.get("LANGFUSE_SECRET_KEY"):
-                        logger.warning(
-                            "LANGFUSE_SECRET_KEY is set but langfuse is not installed; "
-                            "install with `pip install langfuse` to enable tracing"
-                        )
-                    from openai import AsyncOpenAI as _AsyncOpenAI
-                AsyncOpenAI = _AsyncOpenAI
-
-            self._build_client()
-            return self._client
+        # Responses API circuit breaker: skip after repeated failures,
+        # probe again after _RESPONSES_PROBE_INTERVAL_S seconds.
+        self._responses_failures: dict[str, int] = {}
+        self._responses_tripped_at: dict[str, float] = {}
 
     def _setup_env(self, api_key: str, api_base: str | None) -> None:
         """Set environment variables based on provider spec."""
@@ -1211,7 +1182,6 @@ class OpenAICompatProvider(LLMProvider):
         reasoning_effort: str | None = None,
         tool_choice: str | dict[str, Any] | None = None,
     ) -> LLMResponse:
-        await self._ensure_client()
         try:
             if self._should_use_responses_api(model, reasoning_effort):
                 try:
@@ -1253,7 +1223,6 @@ class OpenAICompatProvider(LLMProvider):
         on_thinking_delta: Callable[[str], Awaitable[None]] | None = None,
         on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
     ) -> LLMResponse:
-        await self._ensure_client()
         idle_timeout_s = int(os.environ.get("NANOBOT_STREAM_IDLE_TIMEOUT_S", "90"))
         try:
             if self._should_use_responses_api(model, reasoning_effort):

nanobot/providers/openai_responses/converters.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import json
 from typing import Any
 
+from nanobot.providers.base import LLMProvider
+
 
 def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
     """Convert Chat Completions messages to Responses API input items.
@@ -58,8 +60,10 @@ def convert_messages(messages: list[dict[str, Any]]) -> tuple[str, list[dict[str
 def convert_user_message(content: Any) -> dict[str, Any]:
     """Convert a user message's content to Responses API format.
 
-    Handles plain strings, ``text`` blocks -> ``input_text``, and
-    ``image_url`` blocks -> ``input_image``.
+    Handles plain strings, ``text`` blocks -> ``input_text``,
+    ``image_url`` blocks -> ``input_image``, and ``input_audio`` blocks.
+    ``video_url`` is downgraded to a text placeholder because Codex does
+    not support native video.
     """
     if isinstance(content, str):
         return {"role": "user", "content": [{"type": "input_text", "text": content}]}
@@ -74,6 +78,18 @@ def convert_user_message(content: Any) -> dict[str, Any]:
                 url = (item.get("image_url") or {}).get("url")
                 if url:
                     converted.append({"type": "input_image", "image_url": url, "detail": "auto"})
+            elif item.get("type") == "input_audio":
+                audio_info = item.get("input_audio") or {}
+                audio_data = audio_info.get("data")
+                if audio_data:
+                    converted.append({
+                        "type": "input_audio",
+                        "input_audio": {"data": audio_data, "format": audio_info.get("format", "wav")},
+                    })
+            elif item.get("type") == "video_url":
+                # Codex doesn't support native video → text placeholder
+                placeholder = LLMProvider._media_placeholder("video_url", item)
+                converted.append({"type": "input_text", "text": placeholder["text"]})
         if converted:
             return {"role": "user", "content": converted}
     return {"role": "user", "content": [{"type": "input_text", "text": ""}]}

nanobot/providers/registry.py

@@ -34,7 +34,7 @@ class ProviderSpec:
     display_name: str = ""  # shown in `nanobot status`
 
     # which provider implementation to use
-    # "openai_compat" | "anthropic" | "azure_openai" | "openai_codex" | "github_copilot" | "xai_oauth" | "bedrock"
+    # "openai_compat" | "anthropic" | "azure_openai" | "openai_codex" | "github_copilot" | "bedrock"
     backend: str = "openai_compat"
 
     # extra env vars, e.g. (("ZHIPUAI_API_KEY", "{api_key}"),)
@@ -165,7 +165,7 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         env_extras=(("APIFREE_API_KEY", "{api_key}"),),
         is_gateway=True,
         detect_by_base_keyword="apifree.ai",
-        default_api_base="https://api.apifree.ai/agent/v1",
+        default_api_base="https://api.apifree.ai/v1",
     ),
     # AiHubMix: global gateway, OpenAI-compatible interface.
     # strip_model_prefix=True: doesn't understand "anthropic/claude-3",
@@ -291,18 +291,6 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         is_oauth=True,
         supports_max_completion_tokens=True,
     ),
-    # xAI Grok OAuth: SuperGrok subscription-backed Responses API provider
-    ProviderSpec(
-        name="xai_oauth",
-        keywords=("xai-oauth", "grok-oauth", "x-ai-oauth", "xai-grok-oauth"),
-        env_key="",
-        display_name="xAI Grok OAuth",
-        backend="xai_oauth",
-        default_api_base="https://api.x.ai/v1",
-        strip_model_prefix=True,
-        is_oauth=True,
-        supports_max_completion_tokens=True,
-    ),
     # DeepSeek: OpenAI-compatible at api.deepseek.com
     ProviderSpec(
         name="deepseek",

nanobot/providers/xai_oauth_provider.py

@@ -1,768 +0,0 @@
-"""xAI Grok OAuth credential flow and Responses provider."""
-
-from __future__ import annotations
-
-import asyncio
-import base64
-import json
-import os
-import secrets
-import time
-import webbrowser
-from collections.abc import Awaitable, Callable
-from contextlib import suppress
-from dataclasses import dataclass
-from hashlib import sha256
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
-from threading import Event, Thread
-from typing import Any
-from urllib.parse import parse_qs, urlencode, urlparse
-
-import httpx
-from filelock import FileLock
-
-from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
-from nanobot.providers.openai_responses import consume_sse, convert_messages, convert_tools
-
-DEFAULT_XAI_API_BASE = "https://api.x.ai/v1"
-DEFAULT_XAI_AUTH_ISSUER = "https://auth.x.ai"
-DEFAULT_XAI_DISCOVERY_URL = f"{DEFAULT_XAI_AUTH_ISSUER}/.well-known/openid-configuration"
-DEFAULT_XAI_REDIRECT_URI = "http://127.0.0.1:56121/callback"
-DEFAULT_XAI_CLIENT_ID = "b1a00492-073a-47ea-816f-4c329264a828"
-DEFAULT_XAI_SCOPE = "openid profile email offline_access grok-cli:access api:access"
-
-_SERVICE_NAME = "nanobot.xai_oauth"
-_SECRET_USERNAME = "default"
-_TOKEN_SKEW_SECONDS = 60
-_LOGIN_TIMEOUT_SECONDS = 300
-
-
-@dataclass(frozen=True)
-class XaiOAuthEndpoints:
-    authorization_endpoint: str
-    token_endpoint: str
-
-
-@dataclass(frozen=True)
-class XaiOAuthCredential:
-    access_token: str
-    refresh_token: str = ""
-    expires_at: float | None = None
-    account_id: str | None = None
-    token_type: str = "Bearer"
-    api_base: str = DEFAULT_XAI_API_BASE
-    storage: str = "unknown"
-
-    @property
-    def is_expiring(self) -> bool:
-        return self.expires_at is not None and self.expires_at <= time.time() + _TOKEN_SKEW_SECONDS
-
-
-def _nanobot_home() -> Path:
-    override = os.environ.get("NANOBOT_HOME")
-    if override:
-        return Path(override).expanduser()
-    from nanobot.config.loader import get_config_path
-
-    return get_config_path().parent
-
-
-def _auth_dir() -> Path:
-    return _nanobot_home() / "auth"
-
-
-def get_xai_oauth_metadata_path() -> Path:
-    """Return the non-secret xAI OAuth metadata path."""
-    return _auth_dir() / "xai-oauth.json"
-
-
-def _lock_path() -> Path:
-    return get_xai_oauth_metadata_path().with_suffix(".lock")
-
-
-def _write_private_json(path: Path, payload: dict[str, Any]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    with suppress(OSError):
-        path.parent.chmod(0o700)
-    tmp = path.with_suffix(path.suffix + ".tmp")
-    tmp.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
-    with suppress(OSError):
-        tmp.chmod(0o600)
-    tmp.replace(path)
-    with suppress(OSError):
-        path.chmod(0o600)
-
-
-def _read_json(path: Path) -> dict[str, Any]:
-    return json.loads(path.read_text(encoding="utf-8"))
-
-
-def _keyring_set(tokens: dict[str, Any]) -> bool:
-    try:
-        import keyring  # type: ignore[import-not-found]
-
-        keyring.set_password(_SERVICE_NAME, _SECRET_USERNAME, json.dumps(tokens))
-        return True
-    except Exception:
-        return False
-
-
-def _keyring_get() -> dict[str, Any] | None:
-    try:
-        import keyring  # type: ignore[import-not-found]
-
-        raw = keyring.get_password(_SERVICE_NAME, _SECRET_USERNAME)
-    except Exception:
-        return None
-    if not raw:
-        return None
-    try:
-        payload = json.loads(raw)
-    except json.JSONDecodeError:
-        return None
-    return payload if isinstance(payload, dict) else None
-
-
-def _keyring_delete() -> None:
-    try:
-        import keyring  # type: ignore[import-not-found]
-
-        keyring.delete_password(_SERVICE_NAME, _SECRET_USERNAME)
-    except Exception:
-        pass
-
-
-def _token_payload(credential: XaiOAuthCredential) -> dict[str, Any]:
-    return {
-        "access_token": credential.access_token,
-        "refresh_token": credential.refresh_token,
-        "expires_at": credential.expires_at,
-        "token_type": credential.token_type,
-    }
-
-
-def save_xai_oauth_credential(credential: XaiOAuthCredential) -> XaiOAuthCredential:
-    """Persist xAI OAuth tokens, preferring OS keychain storage."""
-    with FileLock(str(_lock_path())):
-        tokens = _token_payload(credential)
-        metadata: dict[str, Any] = {
-            "provider": "xai_oauth",
-            "api_base": credential.api_base,
-            "account_id": credential.account_id,
-            "expires_at": credential.expires_at,
-            "updated_at": int(time.time()),
-        }
-        if _keyring_set(tokens):
-            metadata["storage"] = "keyring"
-        else:
-            metadata["storage"] = "file"
-            metadata["tokens"] = tokens
-        _write_private_json(get_xai_oauth_metadata_path(), metadata)
-        return XaiOAuthCredential(
-            access_token=credential.access_token,
-            refresh_token=credential.refresh_token,
-            expires_at=credential.expires_at,
-            account_id=credential.account_id,
-            token_type=credential.token_type,
-            api_base=credential.api_base,
-            storage=str(metadata["storage"]),
-        )
-
-
-def load_xai_oauth_credential() -> XaiOAuthCredential | None:
-    """Load xAI OAuth credentials from keyring or the private file fallback."""
-    path = get_xai_oauth_metadata_path()
-    if not path.exists():
-        return None
-    with FileLock(str(_lock_path())):
-        try:
-            metadata = _read_json(path)
-        except (OSError, json.JSONDecodeError):
-            return None
-
-        storage = str(metadata.get("storage") or "file")
-        tokens = _keyring_get() if storage == "keyring" else metadata.get("tokens")
-        if not isinstance(tokens, dict):
-            return None
-        access_token = str(tokens.get("access_token") or "")
-        if not access_token:
-            return None
-
-        return XaiOAuthCredential(
-            access_token=access_token,
-            refresh_token=str(tokens.get("refresh_token") or ""),
-            expires_at=_as_float(tokens.get("expires_at") or metadata.get("expires_at")),
-            account_id=_as_str(metadata.get("account_id")),
-            token_type=str(tokens.get("token_type") or "Bearer"),
-            api_base=str(metadata.get("api_base") or DEFAULT_XAI_API_BASE),
-            storage=storage,
-        )
-
-
-def delete_xai_oauth_credentials() -> list[Path]:
-    """Delete persisted xAI OAuth credentials and return removed local paths."""
-    removed: list[Path] = []
-    path = get_xai_oauth_metadata_path()
-    lock_path = _lock_path()
-    with FileLock(str(lock_path)):
-        _keyring_delete()
-        try:
-            path.unlink()
-            removed.append(path)
-        except FileNotFoundError:
-            pass
-    try:
-        lock_path.unlink()
-    except FileNotFoundError:
-        pass
-    return removed
-
-
-def get_xai_oauth_login_status() -> XaiOAuthCredential | None:
-    return load_xai_oauth_credential()
-
-
-def pkce_challenge(verifier: str) -> str:
-    digest = sha256(verifier.encode("ascii")).digest()
-    return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
-
-
-def _new_pkce_verifier() -> str:
-    return base64.urlsafe_b64encode(secrets.token_bytes(48)).decode("ascii").rstrip("=")
-
-
-def build_xai_authorization_url(
-    endpoints: XaiOAuthEndpoints,
-    *,
-    verifier: str,
-    state: str,
-    nonce: str | None = None,
-    redirect_uri: str = DEFAULT_XAI_REDIRECT_URI,
-) -> str:
-    params = {
-        "response_type": "code",
-        "client_id": DEFAULT_XAI_CLIENT_ID,
-        "redirect_uri": redirect_uri,
-        "scope": DEFAULT_XAI_SCOPE,
-        "code_challenge": pkce_challenge(verifier),
-        "code_challenge_method": "S256",
-        "state": state,
-        "nonce": nonce or secrets.token_urlsafe(16),
-        "plan": "generic",
-        "referrer": "nanobot",
-    }
-    return f"{endpoints.authorization_endpoint}?{urlencode(params)}"
-
-
-def discover_xai_oauth_endpoints() -> XaiOAuthEndpoints:
-    try:
-        with httpx.Client(timeout=20.0, follow_redirects=True, trust_env=True) as client:
-            response = client.get(DEFAULT_XAI_DISCOVERY_URL)
-            response.raise_for_status()
-            payload = response.json()
-    except Exception:
-        payload = {}
-
-    endpoints = XaiOAuthEndpoints(
-        authorization_endpoint=str(
-            payload.get("authorization_endpoint")
-            or f"{DEFAULT_XAI_AUTH_ISSUER}/authorize"
-        ),
-        token_endpoint=str(
-            payload.get("token_endpoint")
-            or f"{DEFAULT_XAI_AUTH_ISSUER}/oauth/token"
-        ),
-    )
-    _validate_xai_endpoint(endpoints.authorization_endpoint, "authorization_endpoint")
-    _validate_xai_endpoint(endpoints.token_endpoint, "token_endpoint")
-    return endpoints
-
-
-def _validate_xai_endpoint(url: str, label: str) -> None:
-    parsed = urlparse(url)
-    host = parsed.hostname or ""
-    if parsed.scheme != "https" or not (host == "x.ai" or host.endswith(".x.ai")):
-        raise RuntimeError(f"Refusing non-xAI OAuth {label}: {url}")
-
-
-def _parse_callback_value(raw: str) -> tuple[str, str | None]:
-    raw = raw.strip()
-    parsed = urlparse(raw)
-    if parsed.scheme and parsed.netloc:
-        params = parse_qs(parsed.query)
-        code = (params.get("code") or [""])[0]
-        state = (params.get("state") or [None])[0]
-        if not code:
-            raise RuntimeError("OAuth callback URL did not contain a code.")
-        return code, state
-    if raw.startswith("?") or "=" in raw:
-        params = parse_qs(raw.lstrip("?"))
-        code = (params.get("code") or [""])[0]
-        state = (params.get("state") or [None])[0]
-        if not code:
-            raise RuntimeError("OAuth callback query did not contain a code.")
-        return code, state
-    if raw:
-        return raw, None
-    raise RuntimeError("No OAuth code provided.")
-
-
-def _decode_jwt_payload(token: str) -> dict[str, Any]:
-    parts = token.split(".")
-    if len(parts) < 2:
-        return {}
-    data = parts[1] + "=" * (-len(parts[1]) % 4)
-    try:
-        decoded = base64.urlsafe_b64decode(data.encode("ascii"))
-        payload = json.loads(decoded)
-    except Exception:
-        return {}
-    return payload if isinstance(payload, dict) else {}
-
-
-def _credential_from_token_response(payload: dict[str, Any], previous: XaiOAuthCredential | None = None) -> XaiOAuthCredential:
-    access_token = str(payload.get("access_token") or "")
-    if not access_token:
-        raise RuntimeError("xAI token response did not include an access token.")
-
-    claims = _decode_jwt_payload(access_token)
-    id_claims = _decode_jwt_payload(str(payload.get("id_token") or ""))
-    expires_at = _as_float(payload.get("expires_at"))
-    if expires_at is None:
-        expires_in = _as_float(payload.get("expires_in"))
-        expires_at = time.time() + expires_in if expires_in else _as_float(claims.get("exp"))
-
-    account_id = (
-        _as_str(id_claims.get("email"))
-        or _as_str(id_claims.get("preferred_username"))
-        or _as_str(id_claims.get("sub"))
-        or _as_str(claims.get("sub"))
-        or (previous.account_id if previous else None)
-    )
-    refresh_token = str(payload.get("refresh_token") or (previous.refresh_token if previous else ""))
-
-    return XaiOAuthCredential(
-        access_token=access_token,
-        refresh_token=refresh_token,
-        expires_at=expires_at,
-        account_id=account_id,
-        token_type=str(payload.get("token_type") or (previous.token_type if previous else "Bearer")),
-        api_base=previous.api_base if previous else DEFAULT_XAI_API_BASE,
-    )
-
-
-def exchange_xai_oauth_code(
-    code: str,
-    *,
-    verifier: str,
-    endpoints: XaiOAuthEndpoints | None = None,
-    redirect_uri: str = DEFAULT_XAI_REDIRECT_URI,
-) -> XaiOAuthCredential:
-    endpoints = endpoints or discover_xai_oauth_endpoints()
-    challenge = pkce_challenge(verifier)
-    with httpx.Client(timeout=30.0, follow_redirects=True, trust_env=True) as client:
-        response = client.post(
-            endpoints.token_endpoint,
-            headers={"Accept": "application/json"},
-            data={
-                "grant_type": "authorization_code",
-                "client_id": DEFAULT_XAI_CLIENT_ID,
-                "code": code,
-                "redirect_uri": redirect_uri,
-                "code_verifier": verifier,
-                "code_challenge": challenge,
-                "code_challenge_method": "S256",
-            },
-        )
-    if response.status_code >= 400:
-        raise RuntimeError(f"xAI token exchange failed: HTTP {response.status_code}: {response.text[:500]}")
-    return _credential_from_token_response(response.json())
-
-
-def refresh_xai_oauth_credential(credential: XaiOAuthCredential | None = None) -> XaiOAuthCredential:
-    credential = credential or load_xai_oauth_credential()
-    if not credential or not credential.refresh_token:
-        raise RuntimeError("xAI Grok OAuth is not logged in. Run: nanobot provider login xai-oauth")
-
-    endpoints = discover_xai_oauth_endpoints()
-    with httpx.Client(timeout=30.0, follow_redirects=True, trust_env=True) as client:
-        response = client.post(
-            endpoints.token_endpoint,
-            headers={"Accept": "application/json"},
-            data={
-                "grant_type": "refresh_token",
-                "client_id": DEFAULT_XAI_CLIENT_ID,
-                "refresh_token": credential.refresh_token,
-            },
-        )
-    if response.status_code >= 400:
-        raise RuntimeError(f"xAI token refresh failed: HTTP {response.status_code}: {response.text[:500]}")
-    return save_xai_oauth_credential(_credential_from_token_response(response.json(), previous=credential))
-
-
-def resolve_xai_oauth_credential(*, force_refresh: bool = False) -> XaiOAuthCredential:
-    credential = load_xai_oauth_credential()
-    if not credential:
-        raise RuntimeError("xAI Grok OAuth is not logged in. Run: nanobot provider login xai-oauth")
-    if force_refresh or credential.is_expiring:
-        credential = refresh_xai_oauth_credential(credential)
-    return credential
-
-
-def login_xai_oauth_interactive(
-    print_fn: Callable[[str], None] | None = None,
-    prompt_fn: Callable[[str], str] | None = None,
-    open_browser: bool = True,
-    manual_paste: bool = False,
-    timeout_seconds: int = _LOGIN_TIMEOUT_SECONDS,
-) -> XaiOAuthCredential:
-    """Run browser PKCE login and persist xAI OAuth credentials."""
-    printer = print_fn or print
-    prompt = prompt_fn or input
-    endpoints = discover_xai_oauth_endpoints()
-    verifier = _new_pkce_verifier()
-    state = secrets.token_urlsafe(24)
-    nonce = secrets.token_urlsafe(24)
-    authorize_url = build_xai_authorization_url(
-        endpoints,
-        verifier=verifier,
-        state=state,
-        nonce=nonce,
-    )
-
-    callback = _LoopbackCallback()
-    server_started = False if manual_paste else callback.start()
-    printer(f"Open: {authorize_url}")
-    if open_browser:
-        with suppress(Exception):
-            webbrowser.open(authorize_url)
-
-    result: dict[str, str] | None = None
-    if manual_paste:
-        printer("Paste the callback URL or xAI fallback code after authorization.")
-    elif server_started:
-        try:
-            result = callback.wait(timeout_seconds)
-        finally:
-            callback.stop()
-    else:
-        printer("Loopback port 56121 is unavailable; paste the callback URL or xAI fallback code.")
-
-    if result:
-        code = result.get("code") or ""
-        returned_state = result.get("state")
-    else:
-        pasted = prompt("Paste callback URL or fallback code")
-        code, returned_state = _parse_callback_value(pasted)
-
-    if not code:
-        raise RuntimeError("OAuth login did not return a code.")
-    if returned_state and returned_state != state:
-        raise RuntimeError("OAuth state mismatch. Please retry login.")
-
-    credential = exchange_xai_oauth_code(code, verifier=verifier, endpoints=endpoints)
-    return save_xai_oauth_credential(credential)
-
-
-class _LoopbackCallback:
-    def __init__(self) -> None:
-        self._event = Event()
-        self._result: dict[str, str] = {}
-        self._server: ThreadingHTTPServer | None = None
-        self._thread: Thread | None = None
-
-    def start(self) -> bool:
-        owner = self
-
-        class Handler(BaseHTTPRequestHandler):
-            def do_GET(self) -> None:  # noqa: N802 - stdlib callback name
-                parsed = urlparse(self.path)
-                params = parse_qs(parsed.query)
-                code = (params.get("code") or [""])[0]
-                state = (params.get("state") or [""])[0]
-                if parsed.path != "/callback" or not code:
-                    self.send_response(404)
-                    self.end_headers()
-                    return
-                owner._result = {"code": code, "state": state}
-                owner._event.set()
-                self.send_response(200)
-                self.send_header("Content-Type", "text/html; charset=utf-8")
-                self.end_headers()
-                self.wfile.write(b"<html><body>nanobot xAI OAuth complete. You may close this tab.</body></html>")
-
-            def log_message(self, format: str, *args: Any) -> None:  # noqa: A002
-                return
-
-        class Server(ThreadingHTTPServer):
-            allow_reuse_address = True
-            daemon_threads = True
-
-        try:
-            self._server = Server(("127.0.0.1", 56121), Handler)
-        except OSError:
-            return False
-        self._thread = Thread(target=self._server.serve_forever, daemon=True)
-        self._thread.start()
-        return True
-
-    def wait(self, timeout_seconds: int) -> dict[str, str] | None:
-        if self._event.wait(timeout_seconds):
-            return dict(self._result)
-        return None
-
-    def stop(self) -> None:
-        if self._server:
-            self._server.shutdown()
-            self._server.server_close()
-        if self._thread:
-            self._thread.join(timeout=1)
-
-
-def _as_float(value: Any) -> float | None:
-    try:
-        return float(value)
-    except (TypeError, ValueError):
-        return None
-
-
-def _as_str(value: Any) -> str | None:
-    return value if isinstance(value, str) and value else None
-
-
-DEFAULT_XAI_MODEL = "xai-oauth/grok-4.3"
-
-
-class XaiOAuthProvider(LLMProvider):
-    """Use a SuperGrok OAuth session to call xAI's Responses API."""
-
-    supports_progress_deltas = True
-
-    def __init__(self, default_model: str = DEFAULT_XAI_MODEL, config: Any | None = None):
-        super().__init__(api_key=None, api_base=DEFAULT_XAI_API_BASE)
-        self.default_model = default_model
-        self.config = config
-
-    async def _call_xai(
-        self,
-        messages: list[dict[str, Any]],
-        tools: list[dict[str, Any]] | None,
-        model: str | None,
-        max_tokens: int,
-        temperature: float,
-        reasoning_effort: str | None,
-        tool_choice: str | dict[str, Any] | None,
-        on_content_delta: Callable[[str], Awaitable[None]] | None = None,
-        on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
-    ) -> LLMResponse:
-        body = _build_xai_responses_body(
-            messages=messages,
-            tools=tools,
-            model=model or self.default_model,
-            max_tokens=max_tokens,
-            temperature=temperature,
-            reasoning_effort=reasoning_effort,
-            tool_choice=tool_choice,
-            hosted_x_search=getattr(self.config, "x_search", None),
-        )
-        try:
-            credential = await asyncio.to_thread(resolve_xai_oauth_credential)
-            try:
-                content, tool_calls, finish_reason = await _request_xai(
-                    credential,
-                    body,
-                    on_content_delta=on_content_delta,
-                    on_tool_call_delta=on_tool_call_delta,
-                )
-            except _XaiHTTPError as exc:
-                if exc.status_code != 401:
-                    raise
-                credential = await asyncio.to_thread(resolve_xai_oauth_credential, force_refresh=True)
-                content, tool_calls, finish_reason = await _request_xai(
-                    credential,
-                    body,
-                    on_content_delta=on_content_delta,
-                    on_tool_call_delta=on_tool_call_delta,
-                )
-            return LLMResponse(content=content, tool_calls=tool_calls, finish_reason=finish_reason)
-        except Exception as exc:
-            msg = f"Error calling xAI Grok OAuth: {exc}"
-            retry_after = getattr(exc, "retry_after", None) or self._extract_retry_after(msg)
-            return LLMResponse(content=msg, finish_reason="error", retry_after=retry_after)
-
-    async def chat(
-        self,
-        messages: list[dict[str, Any]],
-        tools: list[dict[str, Any]] | None = None,
-        model: str | None = None,
-        max_tokens: int = 4096,
-        temperature: float = 0.7,
-        reasoning_effort: str | None = None,
-        tool_choice: str | dict[str, Any] | None = None,
-    ) -> LLMResponse:
-        return await self._call_xai(
-            messages,
-            tools,
-            model,
-            max_tokens,
-            temperature,
-            reasoning_effort,
-            tool_choice,
-        )
-
-    async def chat_stream(
-        self,
-        messages: list[dict[str, Any]],
-        tools: list[dict[str, Any]] | None = None,
-        model: str | None = None,
-        max_tokens: int = 4096,
-        temperature: float = 0.7,
-        reasoning_effort: str | None = None,
-        tool_choice: str | dict[str, Any] | None = None,
-        on_content_delta: Callable[[str], Awaitable[None]] | None = None,
-        on_thinking_delta: Callable[[str], Awaitable[None]] | None = None,
-        on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
-    ) -> LLMResponse:
-        _ = on_thinking_delta
-        return await self._call_xai(
-            messages,
-            tools,
-            model,
-            max_tokens,
-            temperature,
-            reasoning_effort,
-            tool_choice,
-            on_content_delta,
-            on_tool_call_delta,
-        )
-
-    def get_default_model(self) -> str:
-        return self.default_model
-
-
-def _strip_model_prefix(model: str) -> str:
-    for prefix in ("xai-oauth/", "xai_oauth/", "grok-oauth/", "grok_oauth/"):
-        if model.startswith(prefix):
-            return model.split("/", 1)[1]
-    return model
-
-
-def _build_xai_responses_body(
-    *,
-    messages: list[dict[str, Any]],
-    tools: list[dict[str, Any]] | None,
-    model: str,
-    max_tokens: int,
-    temperature: float,
-    reasoning_effort: str | None,
-    tool_choice: str | dict[str, Any] | None,
-    hosted_x_search: Any | None = None,
-) -> dict[str, Any]:
-    system_prompt, input_items = convert_messages(LLMProvider._sanitize_empty_content(messages))
-    if system_prompt:
-        input_items = [
-            {"role": "system", "content": [{"type": "input_text", "text": system_prompt}]},
-            *input_items,
-        ]
-
-    body: dict[str, Any] = {
-        "model": _strip_model_prefix(model),
-        "store": False,
-        "stream": True,
-        "input": input_items,
-        "tool_choice": tool_choice or "auto",
-        "parallel_tool_calls": True,
-    }
-    if max_tokens:
-        body["max_output_tokens"] = max_tokens
-    if temperature is not None:
-        body["temperature"] = temperature
-    if reasoning_effort and reasoning_effort.lower() != "none":
-        body["reasoning"] = {"effort": reasoning_effort}
-    converted_tools = convert_tools(tools) if tools else []
-    hosted_tool = _build_xai_hosted_x_search_tool(hosted_x_search)
-    if hosted_tool:
-        converted_tools.append(hosted_tool)
-    if converted_tools:
-        body["tools"] = converted_tools
-    return body
-
-
-def _clean_x_handles(handles: list[str] | None) -> list[str] | None:
-    if not handles:
-        return None
-    cleaned = [str(handle).strip().lstrip("@") for handle in handles if str(handle).strip()]
-    return cleaned[:10] or None
-
-
-def _build_xai_hosted_x_search_tool(config: Any | None) -> dict[str, Any] | None:
-    if not config or not getattr(config, "enable", False):
-        return None
-
-    allowed = _clean_x_handles(getattr(config, "allowed_x_handles", None))
-    excluded = _clean_x_handles(getattr(config, "excluded_x_handles", None))
-    if allowed and excluded:
-        raise ValueError("providers.xai_oauth.x_search cannot set both allowed_x_handles and excluded_x_handles")
-
-    tool: dict[str, Any] = {"type": "x_search"}
-    if allowed:
-        tool["allowed_x_handles"] = allowed
-    if excluded:
-        tool["excluded_x_handles"] = excluded
-    if getattr(config, "from_date", None):
-        tool["from_date"] = config.from_date
-    if getattr(config, "to_date", None):
-        tool["to_date"] = config.to_date
-    if getattr(config, "enable_image_understanding", False):
-        tool["enable_image_understanding"] = True
-    if getattr(config, "enable_video_understanding", False):
-        tool["enable_video_understanding"] = True
-    return tool
-
-
-class _XaiHTTPError(RuntimeError):
-    def __init__(self, message: str, *, status_code: int, retry_after: float | None = None):
-        super().__init__(message)
-        self.status_code = status_code
-        self.retry_after = retry_after
-
-
-async def _request_xai(
-    credential: XaiOAuthCredential,
-    body: dict[str, Any],
-    on_content_delta: Callable[[str], Awaitable[None]] | None = None,
-    on_tool_call_delta: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
-) -> tuple[str, list[ToolCallRequest], str]:
-    url = credential.api_base.rstrip("/") + "/responses"
-    headers = {
-        "Authorization": f"Bearer {credential.access_token}",
-        "Accept": "text/event-stream",
-        "Content-Type": "application/json",
-        "User-Agent": "nanobot (python)",
-    }
-    timeout = httpx.Timeout(120.0, connect=20.0)
-    async with httpx.AsyncClient(timeout=timeout, follow_redirects=True, trust_env=True) as client:
-        async with client.stream("POST", url, headers=headers, json=body) as response:
-            if response.status_code != 200:
-                raw = await response.aread()
-                retry_after = LLMProvider._extract_retry_after_from_headers(response.headers)
-                raise _XaiHTTPError(
-                    _friendly_error(response.status_code, raw.decode("utf-8", "ignore")),
-                    status_code=response.status_code,
-                    retry_after=retry_after,
-                )
-            return await consume_sse(response, on_content_delta, on_tool_call_delta)
-
-
-def _friendly_error(status_code: int, raw: str) -> str:
-    if status_code == 401:
-        return "xAI OAuth session expired or was revoked. Run: nanobot provider login xai-oauth"
-    if status_code == 403:
-        return (
-            "xAI accepted the OAuth token, but this account is not entitled for the requested "
-            "Grok API capability yet. Check the active Grok subscription and selected model."
-        )
-    if status_code == 429:
-        return "xAI Grok subscription quota or rate limit was reached. Please try again later."
-    return f"HTTP {status_code}: {raw[:500]}"

nanobot/utils/helpers.py

@@ -171,6 +171,79 @@ def detect_image_mime(data: bytes) -> str | None:
     return None
 
 
+# Audio formats supported by OpenAI input_audio block
+_AUDIO_MIME_COMPAT = {"audio/wav", "audio/mpeg", "audio/mp3", "audio/aac",
+                      "audio/ogg", "audio/flac", "audio/x-m4a", "audio/mp4"}
+
+# Map MIME types to the format token expected by OpenAI-compatible input_audio APIs.
+_AUDIO_FORMAT_MAP: dict[str, str] = {
+    "audio/wav": "wav",
+    "audio/x-wav": "wav",
+    "audio/mpeg": "mp3",
+    "audio/mp3": "mp3",
+    "audio/aac": "aac",
+    "audio/ogg": "ogg",
+    "audio/flac": "flac",
+    "audio/x-m4a": "m4a",
+    "audio/mp4": "m4a",
+}
+
+
+def detect_audio_mime(data: bytes, filename: str = "") -> str | None:
+    """Detect audio MIME type from magic bytes; fallback to filename guess."""
+    if data[:4] == b"RIFF" and data[8:12] == b"WAVE":
+        return "audio/wav"
+    if data[:2] in (b"\xff\xfb", b"\xff\xf3", b"\xff\xf2", b"\xff\xfa"):
+        return "audio/mpeg"
+    if data[:4] == b"fLaC":
+        return "audio/flac"
+    if data[:4] == b"OggS":
+        return "audio/ogg"
+    if len(data) > 8 and data[4:8] == b"ftyp":
+        # Only claim audio for M4A-specific brands; avoid matching MP4 video.
+        brand = data[8:12]
+        if brand in (b"M4A ", b"M4AB", b"M4AC"):
+            return "audio/x-m4a"
+    if filename:
+        import mimetypes as _mt
+        guessed = _mt.guess_type(filename)[0]
+        if guessed and guessed.startswith("audio/"):
+            return guessed
+    return None
+
+
+def audio_mime_compat(mime: str | None) -> bool:
+    """Check if the audio MIME is compatible with OpenAI input_audio block."""
+    if not mime:
+        return False
+    return mime in _AUDIO_MIME_COMPAT
+
+
+def audio_format_for_api(mime: str) -> str:
+    """Convert an audio MIME type to the format token expected by the API.
+
+    Falls back to the subtype portion of the MIME (e.g. "x-m4a" from
+    "audio/x-m4a") when no explicit mapping exists.
+    """
+    if not mime:
+        return "wav"
+    return _AUDIO_FORMAT_MAP.get(mime, mime.split("/")[-1])
+
+
+# Video formats commonly supported by LLM APIs (data URI inline)
+_VIDEO_MIME_COMPAT = {
+    "video/mp4", "video/quicktime", "video/x-m4v",
+    "video/webm", "video/x-matroska",
+}
+
+
+def video_mime_compat(mime: str | None) -> bool:
+    """Check if the video MIME is in the commonly-supported set."""
+    if not mime:
+        return False
+    return mime in _VIDEO_MIME_COMPAT
+
+
 def build_image_content_blocks(
     raw: bytes, mime: str, path: str, label: str
 ) -> list[dict[str, Any]]:

pyproject.toml

@@ -61,7 +61,6 @@ dependencies = [
     "openpyxl>=3.1.0,<4.0.0",
     "python-pptx>=1.0.0,<2.0.0",
     "filelock>=3.25.2",
-    "keyring>=25.0.0,<26.0.0",
     "boto3>=1.43.0",
 ]
 

tests/agent/test_multimodal.py

@@ -0,0 +1,178 @@
+from pathlib import Path
+
+from nanobot.agent.context import ContextBuilder
+from nanobot.config.schema import InputLimitsConfig
+from nanobot.utils.helpers import detect_audio_mime, video_mime_compat
+
+PNG_BYTES = (
+    b"\x89PNG\r\n\x1a\n"
+    b"\x00\x00\x00\rIHDR"
+    b"\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00"
+    b"\x90wS\xde"
+    b"\x00\x00\x00\x0cIDATx\x9cc``\x00\x00\x00\x04\x00\x01"
+    b"\x0b\x0e-\xb4"
+    b"\x00\x00\x00\x00IEND\xaeB`\x82"
+)
+
+WAV_BYTES = b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00\x01\x00\x01\x00"
+
+
+def _builder(tmp_path: Path, input_limits: InputLimitsConfig | None = None) -> ContextBuilder:
+    return ContextBuilder(tmp_path, input_limits=input_limits)
+
+
+class TestAudioDetection:
+    def test_detect_wav_from_magic_bytes(self) -> None:
+        assert detect_audio_mime(WAV_BYTES) == "audio/wav"
+
+    def test_detect_mp3_from_magic_bytes(self) -> None:
+        mp3 = b"\xff\xfb\x90\x00"
+        assert detect_audio_mime(mp3) == "audio/mpeg"
+
+    def test_detect_fallback_to_filename(self) -> None:
+        assert detect_audio_mime(b"unknown", filename="song.mp3") == "audio/mpeg"
+
+    def test_returns_none_for_non_audio(self) -> None:
+        assert detect_audio_mime(PNG_BYTES) is None
+
+
+class TestVideoMimeCompat:
+    def test_mp4_is_compatible(self) -> None:
+        assert video_mime_compat("video/mp4") is True
+
+    def test_unknown_is_not_compatible(self) -> None:
+        assert video_mime_compat("video/avi") is False
+
+    def test_none_is_not_compatible(self) -> None:
+        assert video_mime_compat(None) is False
+
+
+class TestBuildUserContentMultimodal:
+    def test_audio_block_when_supported(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        path = tmp_path / "voice.wav"
+        path.write_bytes(WAV_BYTES)
+
+        content = builder._build_user_content("transcribe", [str(path)], supports_audio=True)
+
+        assert isinstance(content, list)
+        audio_blocks = [b for b in content if b.get("type") == "input_audio"]
+        assert len(audio_blocks) == 1
+        assert audio_blocks[0]["input_audio"]["format"] == "wav"
+
+    def test_audio_placeholder_when_not_supported(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        path = tmp_path / "voice.wav"
+        path.write_bytes(WAV_BYTES)
+
+        content = builder._build_user_content("transcribe", [str(path)], supports_audio=False)
+
+        assert isinstance(content, list)
+        assert any("[audio:" in b.get("text", "") for b in content)
+
+    def test_video_block_when_supported(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        path = tmp_path / "clip.mp4"
+        path.write_bytes(b"\x00" * 64)
+
+        content = builder._build_user_content("describe", [str(path)], supports_video=True)
+
+        assert isinstance(content, list)
+        video_blocks = [b for b in content if b.get("type") == "video_url"]
+        assert len(video_blocks) == 1
+        assert video_blocks[0]["video_url"]["url"].startswith("data:video/mp4;base64,")
+
+    def test_video_placeholder_when_not_supported(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        path = tmp_path / "clip.mp4"
+        path.write_bytes(b"\x00" * 64)
+
+        content = builder._build_user_content("describe", [str(path)], supports_video=False)
+
+        assert isinstance(content, list)
+        assert any("[video:" in b.get("text", "") for b in content)
+
+    def test_vision_fallback_downgrades_image(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        path = tmp_path / "pic.png"
+        path.write_bytes(PNG_BYTES)
+
+        content = builder._build_user_content("look", [str(path)], supports_vision=False)
+
+        assert isinstance(content, list)
+        assert any("[image:" in b.get("text", "") for b in content)
+        assert not any(b.get("type") == "image_url" for b in content)
+
+    def test_image_limit_count(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        paths = []
+        for i in range(5):
+            path = tmp_path / f"img{i}.png"
+            path.write_bytes(PNG_BYTES)
+            paths.append(str(path))
+
+        content = builder._build_user_content("describe", paths)
+
+        assert isinstance(content, list)
+        image_count = sum(1 for b in content if b.get("type") == "image_url")
+        assert image_count == 3  # default max_input_images
+        assert any("only the first 3 images" in b.get("text", "") for b in content)
+
+    def test_image_limit_bytes(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        big = tmp_path / "big.png"
+        big.write_bytes(PNG_BYTES + b"x" * builder.input_limits.max_input_image_bytes)
+
+        content = builder._build_user_content("analyze", [str(big)])
+
+        assert isinstance(content, str)
+        assert "file too large" in content
+
+    def test_audio_limit_count(self, tmp_path: Path) -> None:
+        limits = InputLimitsConfig(max_input_audios=1)
+        builder = _builder(tmp_path, input_limits=limits)
+        for i in range(2):
+            path = tmp_path / f"snd{i}.wav"
+            path.write_bytes(WAV_BYTES)
+
+        content = builder._build_user_content(
+            "compare", [str(tmp_path / "snd0.wav"), str(tmp_path / "snd1.wav")], supports_audio=True
+        )
+
+        assert isinstance(content, list)
+        audio_count = sum(1 for b in content if b.get("type") == "input_audio")
+        assert audio_count == 1
+        assert any("only 1 audio" in b.get("text", "") for b in content)
+
+    def test_audio_limit_bytes(self, tmp_path: Path) -> None:
+        limits = InputLimitsConfig(max_input_audio_bytes=32)
+        builder = _builder(tmp_path, input_limits=limits)
+        path = tmp_path / "big.wav"
+        path.write_bytes(WAV_BYTES + b"x" * 64)
+
+        content = builder._build_user_content("analyze", [str(path)], supports_audio=True)
+
+        assert isinstance(content, str)
+        assert "file too large" in content
+
+    def test_mixed_media_types(self, tmp_path: Path) -> None:
+        builder = _builder(tmp_path)
+        img = tmp_path / "pic.png"
+        img.write_bytes(PNG_BYTES)
+        snd = tmp_path / "voice.wav"
+        snd.write_bytes(WAV_BYTES)
+        vid = tmp_path / "clip.mp4"
+        vid.write_bytes(b"\x00" * 64)
+
+        content = builder._build_user_content(
+            "analyze",
+            [str(img), str(snd), str(vid)],
+            supports_vision=True,
+            supports_audio=True,
+            supports_video=True,
+        )
+
+        assert isinstance(content, list)
+        assert any(b.get("type") == "image_url" for b in content)
+        assert any(b.get("type") == "input_audio" for b in content)
+        assert any(b.get("type") == "video_url" for b in content)

tests/channels/test_channel_plugins.py

@@ -111,23 +111,6 @@ def test_discover_plugins_loads_entry_points():
     assert result["line"] is _FakePlugin
 
 
-def test_discover_plugins_skips_names_outside_enabled_set():
-    from nanobot.channels.registry import discover_plugins
-
-    loaded: list[str] = []
-
-    def _load_disabled():
-        loaded.append("disabled")
-        return _FakePlugin
-
-    ep = SimpleNamespace(name="disabled", load=_load_disabled)
-    with patch(_EP_TARGET, return_value=[ep]):
-        result = discover_plugins({"enabled"})
-
-    assert result == {}
-    assert loaded == []
-
-
 def test_discover_plugins_handles_load_error():
     from nanobot.channels.registry import discover_plugins
 
@@ -169,25 +152,6 @@ def test_discover_all_includes_external_plugin():
     assert result["line"] is _FakePlugin
 
 
-def test_discover_enabled_imports_only_enabled_builtins():
-    from nanobot.channels.registry import discover_enabled
-
-    loaded: list[str] = []
-
-    def _load_channel(name: str):
-        loaded.append(name)
-        return _FakePlugin
-
-    with (
-        patch("nanobot.channels.registry.load_channel_class", side_effect=_load_channel),
-        patch(_EP_TARGET, return_value=[]),
-    ):
-        result = discover_enabled({"enabled"}, _names=["enabled", "disabled"])
-
-    assert result == {"enabled": _FakePlugin}
-    assert loaded == ["enabled"]
-
-
 def test_discover_all_builtin_shadows_plugin():
     from nanobot.channels.registry import discover_all
 
@@ -216,7 +180,7 @@ async def test_manager_loads_plugin_from_dict_config():
     )
 
     with patch(
-        "nanobot.channels.registry.discover_enabled",
+        "nanobot.channels.registry.discover_all",
         return_value={"fakeplugin": _FakePlugin},
     ):
         mgr = ChannelManager.__new__(ChannelManager)
@@ -246,7 +210,7 @@ async def test_manager_propagates_groq_transcription_api_base_to_channels():
     )
 
     with patch(
-        "nanobot.channels.registry.discover_enabled",
+        "nanobot.channels.registry.discover_all",
         return_value={"fakeplugin": _FakePlugin},
     ):
         mgr = ChannelManager.__new__(ChannelManager)
@@ -282,7 +246,7 @@ async def test_manager_propagates_openai_transcription_api_base_to_channels():
     )
 
     with patch(
-        "nanobot.channels.registry.discover_enabled",
+        "nanobot.channels.registry.discover_all",
         return_value={"fakeplugin": _FakePlugin},
     ):
         mgr = ChannelManager.__new__(ChannelManager)
@@ -534,8 +498,10 @@ async def test_manager_skips_disabled_plugin():
         providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
     )
 
-    ep = _make_entry_point("fakeplugin", _FakePlugin)
-    with patch(_EP_TARGET, return_value=[ep]):
+    with patch(
+        "nanobot.channels.registry.discover_all",
+        return_value={"fakeplugin": _FakePlugin},
+    ):
         mgr = ChannelManager.__new__(ChannelManager)
         mgr.config = fake_config
         mgr.bus = MessageBus()

tests/channels/test_feishu_post_content.py

@@ -27,10 +27,11 @@ def test_extract_post_content_supports_post_wrapper_shape() -> None:
         }
     }
 
-    text, image_keys = _extract_post_content(payload)
+    text, image_keys, media_items = _extract_post_content(payload)
 
     assert text == "日报 完成"
     assert image_keys == ["img_1"]
+    assert media_items == []
 
 
 def test_extract_post_content_keeps_direct_shape_behavior() -> None:
@@ -45,10 +46,29 @@ def test_extract_post_content_keeps_direct_shape_behavior() -> None:
         ],
     }
 
-    text, image_keys = _extract_post_content(payload)
+    text, image_keys, media_items = _extract_post_content(payload)
 
     assert text == "Daily report"
     assert image_keys == ["img_a", "img_b"]
+    assert media_items == []
+
+
+def test_extract_post_content_extracts_media_tags() -> None:
+    payload = {
+        "title": "Video",
+        "content": [
+            [
+                {"tag": "text", "text": "see this"},
+                {"tag": "media", "file_key": "vid_1"},
+            ]
+        ],
+    }
+
+    text, image_keys, media_items = _extract_post_content(payload)
+
+    assert text == "Video see this"
+    assert image_keys == []
+    assert media_items == [{"tag": "media", "file_key": "vid_1"}]
 
 
 def test_register_optional_event_keeps_builder_when_method_missing() -> None:

tests/channels/test_websocket_channel.py

@@ -1031,7 +1031,7 @@ async def test_settings_api_returns_safe_subset_and_updates_whitelist(
         assert providers["openrouter"]["configured"] is False
         assert providers["openrouter"]["api_key_required"] is True
         assert providers["skywork"]["label"] == "Skywork"
-        assert providers["skywork"]["default_api_base"] == "https://api.apifree.ai/agent/v1"
+        assert providers["skywork"]["default_api_base"] == "https://api.apifree.ai/v1"
         assert providers["ant_ling"]["label"] == "Ant Ling"
         assert providers["ant_ling"]["default_api_base"] == "https://api.ant-ling.com/v1"
         assert providers["atomic_chat"]["configured"] is False

tests/cli/test_commands.py

@@ -11,7 +11,7 @@ from typer.testing import CliRunner
 from nanobot.bus.events import OutboundMessage
 from nanobot.cli.commands import app
 from nanobot.providers.factory import make_provider
-from nanobot.config.schema import Config, ModelPresetConfig
+from nanobot.config.schema import Config
 from nanobot.cron.types import CronJob, CronPayload
 from nanobot.providers.factory import ProviderSnapshot
 from nanobot.providers.openai_codex_provider import _strip_model_prefix
@@ -226,16 +226,6 @@ def test_config_dump_excludes_oauth_provider_blocks():
 
     assert "openaiCodex" not in providers
     assert "githubCopilot" not in providers
-    assert "xaiOauth" not in providers
-
-
-def test_config_dump_includes_xai_oauth_when_hosted_search_is_disabled():
-    config = Config()
-    config.providers.xai_oauth.x_search.enable = False
-
-    providers = config.model_dump(by_alias=True)["providers"]
-
-    assert providers["xaiOauth"]["xSearch"]["enable"] is False
 
 
 def test_provider_logout_openai_codex_removes_local_oauth_files(tmp_path, monkeypatch):
@@ -290,175 +280,6 @@ def test_provider_logout_github_copilot_succeeds_when_no_local_oauth_file(monkey
     assert "No local OAuth credentials found for GitHub Copilot" in result.stdout
 
 
-def test_provider_logout_xai_oauth_removes_local_oauth_files(tmp_path, monkeypatch):
-    token_path = tmp_path / "auth" / "xai-oauth.json"
-    lock_path = token_path.with_suffix(".lock")
-    token_path.parent.mkdir(parents=True, exist_ok=True)
-    token_path.write_text("{}", encoding="utf-8")
-    lock_path.write_text("", encoding="utf-8")
-    monkeypatch.setenv("NANOBOT_HOME", str(tmp_path))
-    monkeypatch.setattr("nanobot.providers.xai_oauth_provider._keyring_delete", lambda: None)
-
-    result = runner.invoke(app, ["provider", "logout", "xai-oauth"])
-
-    assert result.exit_code == 0
-    assert not token_path.exists()
-    assert not lock_path.exists()
-    assert "Logged out from xAI Grok OAuth" in result.stdout
-
-
-def test_provider_logout_xai_oauth_succeeds_when_no_local_oauth_file(monkeypatch, tmp_path):
-    monkeypatch.setenv("NANOBOT_HOME", str(tmp_path))
-    monkeypatch.setattr("nanobot.providers.xai_oauth_provider._keyring_delete", lambda: None)
-
-    result = runner.invoke(app, ["provider", "logout", "xai-oauth"])
-
-    assert result.exit_code == 0
-    assert "No local OAuth credentials found for xAI Grok OAuth" in result.stdout
-
-
-def test_provider_login_xai_oauth_forwards_manual_options(monkeypatch):
-    from nanobot.providers.xai_oauth_provider import XaiOAuthCredential
-
-    captured: dict[str, object] = {}
-
-    def fake_login_xai_oauth_interactive(**kwargs):
-        captured.update(kwargs)
-        return XaiOAuthCredential(access_token="access", account_id="acct", storage="keyring")
-
-    monkeypatch.setattr(
-        "nanobot.providers.xai_oauth_provider.login_xai_oauth_interactive",
-        fake_login_xai_oauth_interactive,
-    )
-
-    result = runner.invoke(app, ["provider", "login", "xai-oauth", "--no-browser", "--manual-paste"])
-
-    assert result.exit_code == 0
-    assert captured["open_browser"] is False
-    assert captured["manual_paste"] is True
-    assert "Authenticated with xAI Grok OAuth" in result.stdout
-    assert "nanobot config set agents.defaults.provider xai-oauth" in result.stdout
-
-
-def test_config_set_updates_default_model_selection(tmp_path):
-    config_path = tmp_path / "config.json"
-
-    result = runner.invoke(app, [
-        "config",
-        "set",
-        "--config",
-        str(config_path),
-        "agents.defaults.model_preset",
-        "null",
-    ])
-    assert result.exit_code == 0
-
-    result = runner.invoke(app, [
-        "config",
-        "set",
-        "--config",
-        str(config_path),
-        "agents.defaults.provider",
-        "xai-oauth",
-    ])
-    assert result.exit_code == 0
-
-    result = runner.invoke(app, [
-        "config",
-        "set",
-        "--config",
-        str(config_path),
-        "agents.defaults.model",
-        "xai-oauth/grok-4.3",
-    ])
-    assert result.exit_code == 0
-
-    data = json.loads(config_path.read_text(encoding="utf-8"))
-    config = Config.model_validate(data)
-    assert config.agents.defaults.model_preset is None
-    assert config.agents.defaults.provider == "xai-oauth"
-    assert config.agents.defaults.model == "xai-oauth/grok-4.3"
-
-
-def test_config_set_warns_when_model_preset_would_override_selection(tmp_path):
-    config = Config()
-    config.agents.defaults.model_preset = "fast"
-    config.model_presets["fast"] = ModelPresetConfig(
-        provider="openrouter",
-        model="openrouter/openai/gpt-4o-mini",
-    )
-    config_path = tmp_path / "config.json"
-    config_path.write_text(json.dumps(config.model_dump(mode="json", by_alias=True)), encoding="utf-8")
-
-    result = runner.invoke(app, [
-        "config",
-        "set",
-        "--config",
-        str(config_path),
-        "agents.defaults.provider",
-        "xai-oauth",
-    ])
-
-    assert result.exit_code == 0
-    assert "model_preset is set and may override this" in result.stdout
-
-
-def test_config_set_disables_xai_oauth_hosted_search(tmp_path):
-    config_path = tmp_path / "config.json"
-
-    result = runner.invoke(app, [
-        "config",
-        "set",
-        "--config",
-        str(config_path),
-        "providers.xai_oauth.x_search.enable",
-        "false",
-    ])
-
-    assert result.exit_code == 0
-    data = json.loads(config_path.read_text(encoding="utf-8"))
-    assert data["providers"]["xaiOauth"]["xSearch"]["enable"] is False
-    assert Config.model_validate(data).providers.xai_oauth.x_search.enable is False
-
-
-def test_config_set_rejects_unknown_path(tmp_path):
-    result = runner.invoke(app, [
-        "config",
-        "set",
-        "--config",
-        str(tmp_path / "config.json"),
-        "agents.defaults.not_a_field",
-        "value",
-    ])
-
-    assert result.exit_code == 1
-    assert "Could not set config value" in result.stdout
-
-
-def test_provider_login_xai_oauth_does_not_update_config(monkeypatch, tmp_path):
-    from nanobot.providers.xai_oauth_provider import XaiOAuthCredential
-
-    config = Config()
-    config.agents.defaults.provider = "auto"
-    config.agents.defaults.model = "anthropic/claude-opus-4-5"
-    config_path = tmp_path / "config.json"
-
-    monkeypatch.setattr(
-        "nanobot.providers.xai_oauth_provider.login_xai_oauth_interactive",
-        lambda **_kwargs: XaiOAuthCredential(access_token="access", account_id="acct", storage="keyring"),
-    )
-    monkeypatch.setattr("nanobot.config.loader.get_config_path", lambda: config_path)
-    monkeypatch.setattr("nanobot.config.loader.load_config", lambda _path=None: config)
-    save_config = MagicMock()
-    monkeypatch.setattr("nanobot.config.loader.save_config", save_config)
-
-    result = runner.invoke(app, ["provider", "login", "xai-oauth"])
-
-    assert result.exit_code == 0
-    save_config.assert_not_called()
-    assert "nanobot config set agents.defaults.model xai-oauth/grok-4.3" in result.stdout
-
-
 def test_provider_logout_rejects_unknown_provider():
     result = runner.invoke(app, ["provider", "logout", "not-a-real-provider"])
 
@@ -577,8 +398,6 @@ def test_find_by_name_accepts_camel_case_and_hyphen_aliases():
     assert find_by_name("volcengineCodingPlan").name == "volcengine_coding_plan"
     assert find_by_name("github-copilot") is not None
     assert find_by_name("github-copilot").name == "github_copilot"
-    assert find_by_name("xai-oauth") is not None
-    assert find_by_name("xai-oauth").name == "xai_oauth"
     assert find_by_name("longcat") is not None
     assert find_by_name("longcat").name == "longcat"
     assert find_by_name("atomic-chat") is not None
@@ -721,23 +540,6 @@ def test_make_provider_uses_github_copilot_backend():
     assert provider.__class__.__name__ == "GitHubCopilotProvider"
 
 
-def test_make_provider_uses_xai_oauth_backend():
-    config = Config.model_validate(
-        {
-            "agents": {
-                "defaults": {
-                    "provider": "xai-oauth",
-                    "model": "xai-oauth/grok-4.3",
-                }
-            }
-        }
-    )
-
-    provider = make_provider(config)
-
-    assert provider.__class__.__name__ == "XaiOAuthProvider"
-
-
 def test_github_copilot_provider_strips_prefixed_model_name():
     from nanobot.providers.github_copilot_provider import GitHubCopilotProvider
 
@@ -770,7 +572,6 @@ async def test_github_copilot_provider_refreshes_client_api_key_before_chat():
 
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI", return_value=mock_client):
         provider = GitHubCopilotProvider(default_model="github-copilot/gpt-4")
-        await provider._ensure_client()
 
     provider._get_copilot_access_token = AsyncMock(return_value="copilot-access-token")
 
@@ -810,8 +611,7 @@ def test_make_provider_passes_extra_headers_to_custom_provider():
     )
 
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai:
-        provider = make_provider(config)
-        asyncio.run(provider._ensure_client())
+        make_provider(config)
 
     kwargs = mock_async_openai.call_args.kwargs
     assert kwargs["api_key"] == "test-key"

tests/providers/test_github_copilot_routing.py

@@ -65,7 +65,6 @@ async def test_github_copilot_does_not_fall_back_from_responses_error():
 
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI", return_value=mock_client):
         provider = GitHubCopilotProvider(default_model="github_copilot/gpt-5.4-mini")
-        await provider._ensure_client()
     provider._get_copilot_access_token = AsyncMock(return_value="copilot-access-token")
 
     response = await provider.chat(

tests/providers/test_litellm_kwargs.py

@@ -449,28 +449,27 @@ def test_gemma_routes_to_gemini_provider() -> None:
     assert "gemma" in spec.keywords
 
 
-async def test_openrouter_sets_default_attribution_headers() -> None:
+def test_openrouter_sets_default_attribution_headers() -> None:
     spec = find_by_name("openrouter")
-    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_client_cls:
-        provider = OpenAICompatProvider(
+    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as MockClient:
+        OpenAICompatProvider(
             api_key="sk-or-test-key",
             api_base="https://openrouter.ai/api/v1",
             default_model="anthropic/claude-sonnet-4-5",
             spec=spec,
         )
-        await provider._ensure_client()
 
-    headers = mock_client_cls.call_args.kwargs["default_headers"]
+    headers = MockClient.call_args.kwargs["default_headers"]
     assert headers["HTTP-Referer"] == "https://github.com/HKUDS/nanobot"
     assert headers["X-OpenRouter-Title"] == "nanobot"
     assert headers["X-OpenRouter-Categories"] == "cli-agent,personal-agent"
     assert "x-session-affinity" in headers
 
 
-async def test_openrouter_user_headers_override_default_attribution() -> None:
+def test_openrouter_user_headers_override_default_attribution() -> None:
     spec = find_by_name("openrouter")
-    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_client_cls:
-        provider = OpenAICompatProvider(
+    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as MockClient:
+        OpenAICompatProvider(
             api_key="sk-or-test-key",
             api_base="https://openrouter.ai/api/v1",
             default_model="anthropic/claude-sonnet-4-5",
@@ -481,9 +480,8 @@ async def test_openrouter_user_headers_override_default_attribution() -> None:
             },
             spec=spec,
         )
-        await provider._ensure_client()
 
-    headers = mock_client_cls.call_args.kwargs["default_headers"]
+    headers = MockClient.call_args.kwargs["default_headers"]
     assert headers["HTTP-Referer"] == "https://nanobot.ai"
     assert headers["X-OpenRouter-Title"] == "Nanobot Pro"
     assert headers["X-OpenRouter-Categories"] == "cli-agent,personal-agent"

tests/providers/test_local_endpoint_detection.py

@@ -85,18 +85,17 @@ class TestIsLocalEndpoint:
 class TestLocalKeepaliveConfig:
     """Verify that local endpoints get keepalive_expiry=0."""
 
-    async def test_local_spec_disables_keepalive(self):
+    def test_local_spec_disables_keepalive(self):
         spec = _make_spec(is_local=True)
         spec.env_key = ""
         spec.default_api_base = "http://localhost:11434/v1"
         provider = OpenAICompatProvider(
             api_key="test", api_base="http://localhost:11434/v1", spec=spec,
         )
-        await provider._ensure_client()
         pool = provider._client._client._transport._pool
         assert pool._keepalive_expiry == 0
 
-    async def test_lan_ip_disables_keepalive(self):
+    def test_lan_ip_disables_keepalive(self):
         """A generic 'openai' spec with a LAN IP should still disable keepalive."""
         spec = _make_spec(is_local=False)
         spec.env_key = ""
@@ -104,18 +103,16 @@ class TestLocalKeepaliveConfig:
         provider = OpenAICompatProvider(
             api_key="test", api_base="http://192.168.8.188:1234/v1", spec=spec,
         )
-        await provider._ensure_client()
         pool = provider._client._client._transport._pool
         assert pool._keepalive_expiry == 0
 
-    async def test_cloud_keeps_default_keepalive(self):
+    def test_cloud_keeps_default_keepalive(self):
         spec = _make_spec(is_local=False)
         spec.env_key = ""
         spec.default_api_base = "https://api.openai.com/v1"
         provider = OpenAICompatProvider(
             api_key="test", api_base=None, spec=spec,
         )
-        await provider._ensure_client()
         pool = provider._client._client._transport._pool
         # Default httpx keepalive is 5.0s
         assert pool._keepalive_expiry == 5.0

tests/providers/test_openai_compat_timeout.py

@@ -8,18 +8,16 @@ def _assert_openai_compat_timeout(timeout) -> None:
     assert timeout == 120.0
 
 
-async def test_openai_compat_provider_defers_sdk_client_until_first_use() -> None:
+def test_openai_compat_provider_sets_sdk_timeout() -> None:
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai:
-        provider = OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
-        mock_async_openai.assert_not_called()
-        await provider._ensure_client()
+        OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
 
     kwargs = mock_async_openai.call_args.kwargs
     _assert_openai_compat_timeout(kwargs["timeout"])
     assert kwargs["http_client"] is None
 
 
-async def test_openai_compat_provider_sets_timeout_on_local_http_client() -> None:
+def test_openai_compat_provider_sets_timeout_on_local_http_client() -> None:
     spec = ProviderSpec(
         name="local",
         keywords=(),
@@ -31,13 +29,11 @@ async def test_openai_compat_provider_sets_timeout_on_local_http_client() -> Non
     with (
         patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai,
         patch(
-            "httpx.AsyncClient",
+            "nanobot.providers.openai_compat_provider.httpx.AsyncClient",
             return_value=sentinel.http_client,
         ) as mock_http_client,
     ):
-        provider = OpenAICompatProvider(spec=spec)
-        mock_async_openai.assert_not_called()
-        await provider._ensure_client()
+        OpenAICompatProvider(spec=spec)
 
     client_kwargs = mock_http_client.call_args.kwargs
     _assert_openai_compat_timeout(client_kwargs["timeout"])
@@ -48,11 +44,10 @@ async def test_openai_compat_provider_sets_timeout_on_local_http_client() -> Non
     assert openai_kwargs["http_client"] is sentinel.http_client
 
 
-async def test_openai_compat_provider_timeout_can_be_overridden_by_env(monkeypatch) -> None:
+def test_openai_compat_provider_timeout_can_be_overridden_by_env(monkeypatch) -> None:
     monkeypatch.setenv("NANOBOT_OPENAI_COMPAT_TIMEOUT_S", "45")
 
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai:
-        provider = OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
-        await provider._ensure_client()
+        OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
 
     assert mock_async_openai.call_args.kwargs["timeout"] == 45.0

tests/providers/test_provider_retry.py

@@ -242,7 +242,7 @@ async def test_image_fallback_returns_error_on_second_failure() -> None:
 
 @pytest.mark.asyncio
 async def test_image_fallback_without_meta_uses_default_placeholder() -> None:
-    """When _meta is absent, fallback placeholder is '[image omitted]'."""
+    """When _meta is absent, fallback placeholder is '[image]'."""
     provider = ScriptedProvider([
         LLMResponse(content="error", finish_reason="error"),
         LLMResponse(content="ok"),
@@ -256,7 +256,7 @@ async def test_image_fallback_without_meta_uses_default_placeholder() -> None:
     for msg in msgs_on_retry:
         content = msg.get("content")
         if isinstance(content, list):
-            assert any("[image omitted]" in (b.get("text") or "") for b in content)
+            assert any("[image]" in (b.get("text") or "") for b in content)
 
 
 @pytest.mark.asyncio

tests/providers/test_provider_sdk_retry_defaults.py

@@ -5,10 +5,9 @@ from nanobot.providers.azure_openai_provider import AzureOpenAIProvider
 from nanobot.providers.openai_compat_provider import OpenAICompatProvider
 
 
-async def test_openai_compat_disables_sdk_retries_by_default() -> None:
+def test_openai_compat_disables_sdk_retries_by_default() -> None:
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_client:
-        provider = OpenAICompatProvider(api_key="sk-test", default_model="gpt-4o")
-        await provider._ensure_client()
+        OpenAICompatProvider(api_key="sk-test", default_model="gpt-4o")
 
     kwargs = mock_client.call_args.kwargs
     assert kwargs["max_retries"] == 0

tests/providers/test_providers_init.py

@@ -12,7 +12,6 @@ def test_importing_providers_package_is_lazy(monkeypatch) -> None:
     monkeypatch.delitem(sys.modules, "nanobot.providers.openai_compat_provider", raising=False)
     monkeypatch.delitem(sys.modules, "nanobot.providers.openai_codex_provider", raising=False)
     monkeypatch.delitem(sys.modules, "nanobot.providers.github_copilot_provider", raising=False)
-    monkeypatch.delitem(sys.modules, "nanobot.providers.xai_oauth_provider", raising=False)
     monkeypatch.delitem(sys.modules, "nanobot.providers.azure_openai_provider", raising=False)
     monkeypatch.delitem(sys.modules, "nanobot.providers.bedrock_provider", raising=False)
 
@@ -22,7 +21,6 @@ def test_importing_providers_package_is_lazy(monkeypatch) -> None:
     assert "nanobot.providers.openai_compat_provider" not in sys.modules
     assert "nanobot.providers.openai_codex_provider" not in sys.modules
     assert "nanobot.providers.github_copilot_provider" not in sys.modules
-    assert "nanobot.providers.xai_oauth_provider" not in sys.modules
     assert "nanobot.providers.azure_openai_provider" not in sys.modules
     assert "nanobot.providers.bedrock_provider" not in sys.modules
     assert providers.__all__ == [
@@ -32,7 +30,6 @@ def test_importing_providers_package_is_lazy(monkeypatch) -> None:
         "OpenAICompatProvider",
         "OpenAICodexProvider",
         "GitHubCopilotProvider",
-        "XaiOAuthProvider",
         "AzureOpenAIProvider",
         "BedrockProvider",
     ]
@@ -53,9 +50,3 @@ def test_openai_codex_supports_progress_deltas() -> None:
     from nanobot.providers.openai_codex_provider import OpenAICodexProvider
 
     assert OpenAICodexProvider.supports_progress_deltas is True
-
-
-def test_xai_oauth_supports_progress_deltas() -> None:
-    from nanobot.providers.xai_oauth_provider import XaiOAuthProvider
-
-    assert XaiOAuthProvider.supports_progress_deltas is True

tests/providers/test_skywork_provider.py

@@ -24,7 +24,7 @@ def test_skywork_provider_in_registry() -> None:
     assert skywork.display_name == "Skywork"
     assert skywork.is_gateway is True
     assert skywork.detect_by_base_keyword == "apifree.ai"
-    assert skywork.default_api_base == "https://api.apifree.ai/agent/v1"
+    assert skywork.default_api_base == "https://api.apifree.ai/v1"
     assert skywork.supports_max_completion_tokens is False
 
 
@@ -53,7 +53,7 @@ def test_skywork_model_auto_matches_with_default_api_base() -> None:
 
     assert config.get_provider_name("skywork-ai/skyclaw-v1") == "skywork"
     assert config.get_api_key("skywork-ai/skyclaw-v1") == "sky-key"
-    assert config.get_api_base("skywork-ai/skyclaw-v1") == "https://api.apifree.ai/agent/v1"
+    assert config.get_api_base("skywork-ai/skyclaw-v1") == "https://api.apifree.ai/v1"
 
 
 def test_skywork_preserves_model_id_and_uses_chat_completion_max_tokens() -> None:

tests/providers/test_xai_oauth_auth.py

@@ -1,146 +0,0 @@
-from __future__ import annotations
-
-import json
-import os
-import stat
-from urllib.parse import parse_qs, urlparse
-
-import pytest
-
-import nanobot.providers.xai_oauth_provider as auth
-
-
-def test_build_xai_authorization_url_includes_pkce_and_grok_scope() -> None:
-    endpoints = auth.XaiOAuthEndpoints(
-        authorization_endpoint="https://auth.x.ai/authorize",
-        token_endpoint="https://auth.x.ai/oauth/token",
-    )
-
-    url = auth.build_xai_authorization_url(
-        endpoints,
-        verifier="verifier",
-        state="state",
-        nonce="nonce",
-    )
-
-    parsed = urlparse(url)
-    params = parse_qs(parsed.query)
-    assert parsed.scheme == "https"
-    assert parsed.hostname == "auth.x.ai"
-    assert params["client_id"] == [auth.DEFAULT_XAI_CLIENT_ID]
-    assert params["code_challenge"] == [auth.pkce_challenge("verifier")]
-    assert params["code_challenge_method"] == ["S256"]
-    assert params["scope"] == [auth.DEFAULT_XAI_SCOPE]
-    assert params["nonce"] == ["nonce"]
-    assert params["plan"] == ["generic"]
-    assert params["referrer"] == ["nanobot"]
-
-
-def test_parse_callback_value_accepts_fallback_shapes() -> None:
-    assert auth._parse_callback_value("https://localhost/callback?code=abc&state=state") == ("abc", "state")
-    assert auth._parse_callback_value("?code=abc&state=state") == ("abc", "state")
-    assert auth._parse_callback_value("code=abc&state=state") == ("abc", "state")
-    assert auth._parse_callback_value("fallback-code") == ("fallback-code", None)
-
-
-def test_file_storage_fallback_is_private_and_round_trips(tmp_path, monkeypatch) -> None:
-    monkeypatch.setenv("NANOBOT_HOME", str(tmp_path))
-    monkeypatch.setattr(auth, "_keyring_set", lambda _tokens: False)
-    monkeypatch.setattr(auth, "_keyring_get", lambda: None)
-
-    saved = auth.save_xai_oauth_credential(
-        auth.XaiOAuthCredential(
-            access_token="access",
-            refresh_token="refresh",
-            expires_at=123.0,
-            account_id="acct",
-        )
-    )
-
-    path = auth.get_xai_oauth_metadata_path()
-    payload = json.loads(path.read_text(encoding="utf-8"))
-    assert saved.storage == "file"
-    assert payload["storage"] == "file"
-    assert payload["tokens"]["access_token"] == "access"
-    if os.name != "nt":
-        assert stat.S_IMODE(path.stat().st_mode) == 0o600
-
-    loaded = auth.load_xai_oauth_credential()
-    assert loaded is not None
-    assert loaded.access_token == "access"
-    assert loaded.refresh_token == "refresh"
-    assert loaded.account_id == "acct"
-    assert loaded.storage == "file"
-
-
-def test_keyring_storage_keeps_tokens_out_of_metadata(tmp_path, monkeypatch) -> None:
-    monkeypatch.setenv("NANOBOT_HOME", str(tmp_path))
-    secret: dict[str, object] = {}
-
-    def fake_set(tokens: dict[str, object]) -> bool:
-        secret.update(tokens)
-        return True
-
-    monkeypatch.setattr(auth, "_keyring_set", fake_set)
-    monkeypatch.setattr(auth, "_keyring_get", lambda: dict(secret))
-
-    auth.save_xai_oauth_credential(
-        auth.XaiOAuthCredential(
-            access_token="access",
-            refresh_token="refresh",
-            expires_at=123.0,
-            account_id="acct",
-        )
-    )
-
-    payload = json.loads(auth.get_xai_oauth_metadata_path().read_text(encoding="utf-8"))
-    assert payload["storage"] == "keyring"
-    assert "tokens" not in payload
-    assert auth.load_xai_oauth_credential().access_token == "access"
-
-
-def test_exchange_xai_oauth_code_sends_required_code_challenge(monkeypatch) -> None:
-    captured: dict[str, object] = {}
-
-    class FakeResponse:
-        status_code = 200
-        text = ""
-
-        def json(self) -> dict[str, object]:
-            return {"access_token": "access", "refresh_token": "refresh", "expires_in": 3600}
-
-    class FakeClient:
-        def __init__(self, *args, **kwargs) -> None:
-            pass
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *args) -> None:
-            pass
-
-        def post(self, url: str, headers: dict[str, str], data: dict[str, str]) -> FakeResponse:
-            captured["url"] = url
-            captured["headers"] = headers
-            captured["data"] = data
-            return FakeResponse()
-
-    monkeypatch.setattr(auth.httpx, "Client", FakeClient)
-    endpoints = auth.XaiOAuthEndpoints(
-        authorization_endpoint="https://auth.x.ai/authorize",
-        token_endpoint="https://auth.x.ai/oauth/token",
-    )
-
-    credential = auth.exchange_xai_oauth_code("code", verifier="verifier", endpoints=endpoints)
-
-    assert credential.access_token == "access"
-    assert captured["url"] == "https://auth.x.ai/oauth/token"
-    data = captured["data"]
-    assert data["code_verifier"] == "verifier"
-    assert data["code_challenge"] == auth.pkce_challenge("verifier")
-    assert data["code_challenge_method"] == "S256"
-
-
-def test_rejects_non_xai_discovery_endpoints() -> None:
-    with pytest.raises(RuntimeError):
-        auth._validate_xai_endpoint("https://example.com/oauth/token", "token_endpoint")

tests/providers/test_xai_oauth_provider.py

@@ -1,141 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-
-from nanobot.config.schema import XaiOAuthXSearchConfig
-import nanobot.providers.xai_oauth_provider as xai_oauth_provider
-from nanobot.providers.xai_oauth_provider import (
-    XaiOAuthCredential,
-    XaiOAuthProvider,
-    _build_xai_responses_body,
-    _strip_model_prefix,
-)
-
-
-def test_xai_oauth_strip_prefix_supports_aliases() -> None:
-    assert _strip_model_prefix("xai-oauth/grok-4.3") == "grok-4.3"
-    assert _strip_model_prefix("xai_oauth/grok-4.3") == "grok-4.3"
-    assert _strip_model_prefix("grok-oauth/grok-4.3") == "grok-4.3"
-    assert _strip_model_prefix("grok-4.3") == "grok-4.3"
-
-
-def test_build_xai_responses_body_keeps_system_prompt_in_input() -> None:
-    body = _build_xai_responses_body(
-        messages=[
-            {"role": "system", "content": "You are nanobot."},
-            {"role": "user", "content": "hi"},
-        ],
-        tools=[
-            {
-                "type": "function",
-                "function": {
-                    "name": "ping",
-                    "description": "Ping",
-                    "parameters": {"type": "object", "properties": {}},
-                },
-            }
-        ],
-        model="xai-oauth/grok-4.3",
-        max_tokens=32,
-        temperature=0.2,
-        reasoning_effort="high",
-        tool_choice=None,
-    )
-
-    assert body["model"] == "grok-4.3"
-    assert "instructions" not in body
-    assert body["input"][0] == {
-        "role": "system",
-        "content": [{"type": "input_text", "text": "You are nanobot."}],
-    }
-    assert body["input"][1]["role"] == "user"
-    assert body["max_output_tokens"] == 32
-    assert body["temperature"] == 0.2
-    assert body["reasoning"] == {"effort": "high"}
-    assert body["tools"][0]["name"] == "ping"
-
-
-def test_build_xai_responses_body_attaches_hosted_x_search_by_default() -> None:
-    body = _build_xai_responses_body(
-        messages=[{"role": "user", "content": "what is happening on X?"}],
-        tools=None,
-        model="xai-oauth/grok-4.3",
-        max_tokens=32,
-        temperature=0.2,
-        reasoning_effort=None,
-        tool_choice=None,
-        hosted_x_search=XaiOAuthXSearchConfig(),
-    )
-
-    assert body["tools"] == [{"type": "x_search"}]
-
-
-def test_build_xai_responses_body_can_customize_hosted_x_search() -> None:
-    body = _build_xai_responses_body(
-        messages=[{"role": "user", "content": "what is happening on X?"}],
-        tools=None,
-        model="xai-oauth/grok-4.3",
-        max_tokens=32,
-        temperature=0.2,
-        reasoning_effort=None,
-        tool_choice=None,
-        hosted_x_search=XaiOAuthXSearchConfig(
-            allowed_x_handles=["@xai", " nanobot "],
-            enable_image_understanding=True,
-        ),
-    )
-
-    assert body["tools"] == [
-        {
-            "type": "x_search",
-            "allowed_x_handles": ["xai", "nanobot"],
-            "enable_image_understanding": True,
-        }
-    ]
-
-
-def test_build_xai_responses_body_omits_disabled_hosted_x_search() -> None:
-    body = _build_xai_responses_body(
-        messages=[{"role": "user", "content": "hi"}],
-        tools=None,
-        model="xai-oauth/grok-4.3",
-        max_tokens=32,
-        temperature=0.2,
-        reasoning_effort=None,
-        tool_choice=None,
-        hosted_x_search=XaiOAuthXSearchConfig(enable=False),
-    )
-
-    assert "tools" not in body
-
-
-def test_xai_oauth_provider_refreshes_once_on_401(monkeypatch) -> None:
-    async def run() -> None:
-        response = await provider.chat([{"role": "user", "content": "hi"}])
-
-        assert response.content == "ok"
-        assert response.finish_reason == "stop"
-        assert calls == [("resolve", False), ("resolve", True)]
-
-    provider = XaiOAuthProvider(default_model="xai-oauth/grok-4.3")
-    credentials = [
-        XaiOAuthCredential(access_token="expired"),
-        XaiOAuthCredential(access_token="fresh"),
-    ]
-    calls: list[tuple[str, bool]] = []
-
-    def fake_resolve(*, force_refresh: bool = False) -> XaiOAuthCredential:
-        calls.append(("resolve", force_refresh))
-        return credentials.pop(0)
-
-    async def fake_request(credential, body, on_content_delta=None, on_tool_call_delta=None):
-        from nanobot.providers.xai_oauth_provider import _XaiHTTPError
-
-        if credential.access_token == "expired":
-            raise _XaiHTTPError("expired", status_code=401)
-        return "ok", [], "stop"
-
-    monkeypatch.setattr(xai_oauth_provider, "resolve_xai_oauth_credential", fake_resolve)
-    monkeypatch.setattr(xai_oauth_provider, "_request_xai", fake_request)
-
-    asyncio.run(run())

tests/test_truncate_text_shadowing.py

@@ -29,3 +29,47 @@ def test_sanitize_persisted_blocks_truncate_text_shadowing_regression() -> None:
     assert isinstance(out[0]["text"], str)
     assert out[0]["text"] != content[0]["text"]
 
+
+def test_sanitize_persisted_blocks_strips_audio_and_video() -> None:
+    """Audio and video blocks with base64 payloads must be replaced with placeholders."""
+    from nanobot.agent.loop import AgentLoop
+
+    dummy = SimpleNamespace(max_tool_result_chars=1000)
+    content = [
+        {"type": "text", "text": "analyze this"},
+        {
+            "type": "input_audio",
+            "input_audio": {"data": "aGVsbG8=", "format": "wav"},
+            "_meta": {"path": "/tmp/voice.wav"},
+        },
+        {
+            "type": "video_url",
+            "video_url": {"url": "data:video/mp4;base64,aGVsbG8="},
+            "_meta": {"path": "/tmp/clip.mp4"},
+        },
+    ]
+
+    out = AgentLoop._sanitize_persisted_blocks(dummy, content)
+
+    assert len(out) == 3
+    assert out[0] == content[0]
+    assert out[1] == {"type": "text", "text": "[audio: /tmp/voice.wav]"}
+    assert out[2] == {"type": "text", "text": "[video: /tmp/clip.mp4]"}
+
+
+def test_sanitize_persisted_blocks_strips_audio_video_without_meta() -> None:
+    """When _meta is absent, fallback placeholders use bare label."""
+    from nanobot.agent.loop import AgentLoop
+
+    dummy = SimpleNamespace(max_tool_result_chars=1000)
+    content = [
+        {"type": "input_audio", "input_audio": {"data": "aGVsbG8=", "format": "wav"}},
+        {"type": "video_url", "video_url": {"url": "data:video/mp4;base64,aGVsbG8="}},
+    ]
+
+    out = AgentLoop._sanitize_persisted_blocks(dummy, content)
+
+    assert len(out) == 2
+    assert out[0] == {"type": "text", "text": "[audio]"}
+    assert out[1] == {"type": "text", "text": "[video]"}
+

tests/tools/test_exec_platform.py

@@ -5,7 +5,6 @@ strategy, and sandbox behaviour per platform — without actually running
 platform-specific binaries (all subprocess calls are mocked).
 """
 
-import asyncio
 import sys
 from unittest.mock import AsyncMock, patch
 
@@ -109,9 +108,6 @@ class TestSpawnUnix:
         assert "-c" in args
         assert "echo hi" in args
 
-        kwargs = mock_exec.call_args[1]
-        assert kwargs["stdin"] == asyncio.subprocess.DEVNULL
-
 
 class TestSpawnWindows:
 
@@ -128,9 +124,6 @@ class TestSpawnWindows:
         args = mock_shell.call_args[0]
         assert "dir" in args
 
-        kwargs = mock_shell.call_args[1]
-        assert kwargs["stdin"] == asyncio.subprocess.DEVNULL
-
     @pytest.mark.asyncio
     async def test_passes_cwd_and_env(self):
         env = {"PATH": "/usr/bin"}