feat(providers): add xAI Grok OAuth support

Add APIFree as a built-in OpenAI-compatible provider. APIFree offers
agent-optimised models such as skywork-ai/skyclaw-v1 through an
OpenAI-compatible API at https://api.apifree.ai/agent/v1.

Changes:
- Register apifree provider in the provider registry
- Add config schema field
- Add documentation with configuration example
- Add provider tests, websocket channel tests, and webui tests
- Add provider icon in settings UI

.gitignore

@@ -97,3 +97,4 @@ 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 the OpenAI-compatible APIFree API endpoint. Configure the provider
+Skywork uses APIFree's OpenAI-compatible Agent 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/v1"
+      "apiBase": "https://api.apifree.ai/agent/v1"
     }
   },
   "agents": {

nanobot/__init__.py

@@ -2,9 +2,10 @@
 nanobot - A lightweight AI agent framework
 """
 
-from importlib.metadata import PackageNotFoundError, version as _pkg_version
-from pathlib import Path
 import tomllib
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as _pkg_version
+from pathlib import Path
 
 
 def _read_pyproject_version() -> str | None:
@@ -27,6 +28,21 @@ def _resolve_version() -> str:
 __version__ = _resolve_version()
 __logo__ = "🐈"
 
-from nanobot.nanobot import Nanobot, RunResult
+_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
+
 
 __all__ = ["Nanobot", "RunResult"]

nanobot/agent/context.py

@@ -2,7 +2,6 @@
 
 import base64
 import mimetypes
-import os
 import platform
 from contextlib import suppress
 from importlib.resources import files as pkg_files
@@ -11,16 +10,11 @@ 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
 
@@ -34,12 +28,11 @@ 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, input_limits: InputLimitsConfig | None = None):
+    def __init__(self, workspace: Path, timezone: str | None = None, disabled_skills: list[str] | 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,
@@ -149,28 +142,6 @@ 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]],
@@ -183,9 +154,6 @@ 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)
@@ -196,12 +164,7 @@ class ContextBuilder:
             sender_id=sender_id,
             supplemental_lines=extra or None,
         )
-        user_content = self._build_user_content(
+        user_content = self._build_user_content(current_message, media)
-            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.
@@ -223,171 +186,28 @@ class ContextBuilder:
         messages.append({"role": current_role, "content": merged})
         return messages
 
-    def _build_user_content(
+    def _build_user_content(self, text: str, media: list[str] | None) -> str | list[dict[str, Any]]:
-        self,
+        """Build user message content with optional base64-encoded images."""
-        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
 
-        blocks: list[dict[str, Any]] = []
+        images = []
-        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()
-            # When explicitly marked as non-vision, downgrade to text placeholder
+            mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]
-            if supports_vision is False:
+            if not mime or not mime.startswith("image/"):
-                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)},
+            })
 
-            size_ok = self._file_size_ok(p, limits.max_input_image_bytes)
+        if not images:
-            if size_ok is False:
+            return text
-                size_mb = limits.max_input_image_bytes // (1024 * 1024)
+        return images + [{"type": "text", "text": text}]
-                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,10 +190,6 @@ 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
 
@@ -231,10 +227,6 @@ 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
@@ -248,7 +240,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, input_limits=self.input_limits)
+        self.context = ContextBuilder(workspace, timezone=timezone, disabled_skills=disabled_skills)
         self.sessions = session_manager or SessionManager(workspace)
         self._webui_turns = WebuiTurnCoordinator(
             bus=self.bus,
@@ -374,10 +366,6 @@ 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,
         )
 
@@ -606,9 +594,6 @@ 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(
@@ -1074,9 +1059,6 @@ 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(
@@ -1422,10 +1404,6 @@ 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,6 +266,7 @@ 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,
@@ -274,6 +275,7 @@ 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,22 +172,19 @@ def _extract_element_content(element: dict) -> list[str]:
     return parts
 
 
-def _extract_post_content(content_json: dict) -> tuple[str, list[str], list[dict]]:
+def _extract_post_content(content_json: dict) -> tuple[str, list[str]]:
-    """Extract text and media info from Feishu post (rich text) message.
+    """Extract text and image keys 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], list[dict]]:
+    def _parse_block(block: dict) -> tuple[str | None, list[str]]:
         if not isinstance(block, dict) or not isinstance(block.get("content"), list):
-            return None, [], []
+            return None, []
-        texts, images, medias = [], [], []
+        texts, images = [], []
         if title := block.get("title"):
             texts.append(title)
         for row in block["content"]:
@@ -207,36 +204,43 @@ def _extract_post_content(content_json: dict) -> tuple[str, list[str], list[dict
                     texts.append(f"\n```{lang}\n{code_text}\n```\n")
                 elif tag == "img" and (key := el.get("image_key")):
                     images.append(key)
-                elif tag == "media" and el.get("file_key"):
+        return (" ".join(texts).strip() or None), images
-                    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, medias = _parse_block(root)
+        text, imgs = _parse_block(root)
-        if text or imgs or medias:
+        if text or imgs:
-            return text or "", imgs, medias
+            return text or "", imgs
 
     # 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, medias = _parse_block(root[key])
+            text, imgs = _parse_block(root[key])
-            if text or imgs or medias:
+            if text or imgs:
-                return text or "", imgs, medias
+                return text or "", imgs
     for val in root.values():
         if isinstance(val, dict):
-            text, imgs, medias = _parse_block(val)
+            text, imgs = _parse_block(val)
-            if text or imgs or medias:
+            if text or imgs:
-                return text or "", imgs, medias
+                return text or "", imgs
 
-    return "", [], []
+    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
 
 
 class FeishuConfig(Base):
@@ -1152,7 +1156,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 = ""
@@ -1747,7 +1751,7 @@ class FeishuChannel(BaseChannel):
                     content_parts.append(text)
 
             elif msg_type == "post":
-                text, image_keys, media_items = _extract_post_content(content_json)
+                text, image_keys = _extract_post_content(content_json)
                 if text:
                     content_parts.append(text)
                 # Download images embedded in post
@@ -1758,14 +1762,6 @@ 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,28 +70,40 @@ class ChannelManager:
 
     def _init_channels(self) -> None:
         """Initialize channels discovered via pkgutil scan + entry_points plugins."""
-        from nanobot.channels.registry import discover_all
+        from nanobot.channels.registry import discover_channel_names, discover_enabled
 
         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
 
-        for name, cls in discover_all().items():
+        # 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:
             section = getattr(self.config.channels, name, None)
             if section is None:
                 continue
-            enabled = (
+            if (
                 section.get("enabled", False)
                 if isinstance(section, dict)
                 else getattr(section, "enabled", False)
-            )
+            ):
-            if not enabled:
+                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:
                 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,5 +1,4 @@
 """Auto-discovery for built-in channel modules and external plugins."""
-
 from __future__ import annotations
 
 import importlib
@@ -37,12 +36,14 @@ def load_channel_class(module_name: str) -> type[BaseChannel]:
     raise ImportError(f"No BaseChannel subclass in nanobot.channels.{module_name}")
 
 
-def discover_plugins() -> dict[str, type[BaseChannel]]:
+def discover_plugins(enabled_names: set[str] | None = None) -> 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
@@ -51,21 +52,44 @@ def discover_plugins() -> dict[str, type[BaseChannel]]:
     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.
     """
-    builtin: dict[str, type[BaseChannel]] = {}
+    names = discover_channel_names()
-    for modname in discover_channel_names():
+    return discover_enabled(set(names), _names=names, _include_all_external=True)
-        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,12 +1,14 @@
 """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
 
@@ -1527,6 +1529,106 @@ 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
 # ============================================================================
@@ -1541,6 +1643,7 @@ _LOGOUT_HANDLERS: dict[str, Callable[[], None]] = {}
 _PROVIDER_DISPLAY: dict[str, str] = {
     "openai_codex": "OpenAI Codex",
     "github_copilot": "GitHub Copilot",
+    "xai_oauth": "xAI Grok OAuth",
 }
 
 
@@ -1576,7 +1679,9 @@ 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')"),
+    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."),
 ):
     """Authenticate with an OAuth provider."""
     spec = _resolve_oauth_provider(provider)
@@ -1587,12 +1692,18 @@ def provider_login(
         raise typer.Exit(1)
 
     console.print(f"{__logo__} OAuth Login - {spec.label}\n")
-    handler()
+    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)
 
 
 @provider_app.command("logout")
 def provider_logout(
-    provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot')"),
+    provider: str = typer.Argument(..., help="OAuth provider (e.g. 'openai-codex', 'github-copilot', 'xai-oauth')"),
 ):
     """Log out from an OAuth provider."""
     spec = _resolve_oauth_provider(provider)
@@ -1656,6 +1767,24 @@ 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] = []
@@ -1699,5 +1828,36 @@ 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,35 +155,8 @@ 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."""
@@ -207,6 +180,28 @@ 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."""
 
@@ -244,6 +239,10 @@ 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)
 
@@ -285,17 +284,6 @@ 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
@@ -317,7 +305,6 @@ 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,6 +1,18 @@
 """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,6 +14,7 @@ __all__ = [
     "OpenAICompatProvider",
     "OpenAICodexProvider",
     "GitHubCopilotProvider",
+    "XaiOAuthProvider",
     "AzureOpenAIProvider",
     "BedrockProvider",
 ]
@@ -23,10 +24,23 @@ _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
@@ -34,12 +48,18 @@ 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 None:
+    if module_name is not None:
-        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+        module = import_module(module_name, __name__)
-    module = import_module(module_name, __name__)
+        return getattr(module, 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}")

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 and input_audio blocks."""
+        """Convert user message content, translating image_url blocks."""
         if isinstance(content, str) or content is None:
             return content or "(empty)"
         if not isinstance(content, list):
@@ -228,14 +228,6 @@ 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,6 +13,8 @@ from typing import Any
 
 from loguru import logger
 
+from nanobot.utils.helpers import image_placeholder_text
+
 
 @dataclass
 class ToolCallRequest:
@@ -437,23 +439,9 @@ 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 _media_placeholder(btype: str, block: dict[str, Any]) -> dict[str, str]:
+    def _strip_image_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
-        """Build a text placeholder for a media block."""
+        """Replace image_url blocks with text placeholder. Returns None if no images found."""
-        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:
@@ -461,8 +449,10 @@ class LLMProvider(ABC):
             if isinstance(content, list):
                 new_content = []
                 for b in content:
-                    if isinstance(b, dict) and b.get("type") in LLMProvider._STRIP_MEDIA_TYPES:
+                    if isinstance(b, dict) and b.get("type") == "image_url":
-                        new_content.append(LLMProvider._media_placeholder(b["type"], b))
+                        path = (b.get("_meta") or {}).get("path", "")
+                        placeholder = image_placeholder_text(path, empty="[image omitted]")
+                        new_content.append({"type": "text", "text": placeholder})
                         found = True
                     else:
                         new_content.append(b)
@@ -472,13 +462,8 @@ class LLMProvider(ABC):
         return result if found else None
 
     @staticmethod
-    def _strip_image_content(messages: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
+    def _strip_image_content_inplace(messages: list[dict[str, Any]]) -> bool:
-        """Replace image_url blocks with text placeholder. Returns None if no images found."""
+        """Replace image_url blocks with text placeholder *in-place*.
-        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
@@ -489,16 +474,13 @@ 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") in LLMProvider._STRIP_MEDIA_TYPES:
+                    if isinstance(b, dict) and b.get("type") == "image_url":
-                        content[i] = LLMProvider._media_placeholder(b["type"], b)
+                        path = (b.get("_meta") or {}).get("path", "")
+                        placeholder = image_placeholder_text(path, empty="[image omitted]")
+                        content[i] = {"type": "text", "text": placeholder}
                         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:
@@ -756,18 +738,18 @@ class LLMProvider(ABC):
                 identical_error_count = 1 if error_key else 0
 
             if not self._is_transient_response(response):
-                stripped = self._strip_media_content(original_messages)
+                stripped = self._strip_image_content(original_messages)
                 if stripped is not None and stripped != kw["messages"]:
                     logger.warning(
-                        "Non-transient LLM error with media content, retrying without media"
+                        "Non-transient LLM error with image content, retrying without images"
                     )
                     retry_kw = dict(kw)
                     retry_kw["messages"] = stripped
                     result = await call(**retry_kw)
-                    # Permanently strip media from the original messages so
+                    # Permanently strip images from the original messages so
                     # subsequent iterations do not repeat the error-retry cycle.
                     if result.finish_reason != "error":
-                        self._strip_media_content_inplace(original_messages)
+                        self._strip_image_content_inplace(original_messages)
                     return result
                 return response
 

nanobot/providers/factory.py

@@ -68,6 +68,10 @@ 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,8 +207,9 @@ 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
-        self._client.api_key = token
+        client.api_key = token
         return token
 
     async def chat(

nanobot/providers/openai_compat_provider.py

@@ -16,20 +16,9 @@ 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,
@@ -39,8 +28,15 @@ 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",
@@ -302,43 +298,76 @@ class OpenAICompatProvider(LLMProvider):
 
         effective_base = api_base or (spec.default_api_base if spec else None) or None
         self._effective_base = effective_base
-        default_headers = {"x-session-affinity": uuid.uuid4().hex}
+        self._default_headers = {"x-session-affinity": uuid.uuid4().hex}
         if _uses_openrouter_attribution(spec, effective_base):
-            default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
+            self._default_headers.update(_DEFAULT_OPENROUTER_HEADERS)
         if extra_headers:
-            default_headers.update(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)
 
-        # Local model servers (Ollama, llama.cpp, vLLM) often close idle
+        # Lazy-init: the OpenAI client and its httpx transport are expensive
-        # HTTP connections before the client-side keepalive expires.  When
+        # to create (~700 ms on Windows). Defer until first use.
-        # two LLM calls happen seconds apart (e.g. heartbeat _decide then
+        self._client: AsyncOpenAIType | None = None
-        # process_direct), the second call may grab a now-dead pooled
+        self._client_lock = asyncio.Lock()
-        # 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 _is_local_endpoint(spec, effective_base):
-            http_client = httpx.AsyncClient(
-                limits=httpx.Limits(keepalive_expiry=0),
-                timeout=timeout_s,
-            )
-
-        self._client = AsyncOpenAI(
-            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,
-        )
 
         # 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
+
+        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.
+            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,
+            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
+
     def _setup_env(self, api_key: str, api_base: str | None) -> None:
         """Set environment variables based on provider spec."""
         spec = self._spec
@@ -1182,6 +1211,7 @@ 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:
@@ -1223,6 +1253,7 @@ 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,8 +5,6 @@ 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.
@@ -60,10 +58,8 @@ 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``,
+    Handles plain strings, ``text`` blocks -> ``input_text``, and
-    ``image_url`` blocks -> ``input_image``, and ``input_audio`` blocks.
+    ``image_url`` blocks -> ``input_image``.
-    ``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}]}
@@ -78,18 +74,6 @@ 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" | "bedrock"
+    # "openai_compat" | "anthropic" | "azure_openai" | "openai_codex" | "github_copilot" | "xai_oauth" | "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/v1",
+        default_api_base="https://api.apifree.ai/agent/v1",
     ),
     # AiHubMix: global gateway, OpenAI-compatible interface.
     # strip_model_prefix=True: doesn't understand "anthropic/claude-3",
@@ -291,6 +291,18 @@ 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

@@ -0,0 +1,768 @@
+"""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,79 +171,6 @@ 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,6 +61,7 @@ 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

@@ -1,178 +0,0 @@
-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,6 +111,23 @@ 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
 
@@ -152,6 +169,25 @@ 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
 
@@ -180,7 +216,7 @@ async def test_manager_loads_plugin_from_dict_config():
     )
 
     with patch(
-        "nanobot.channels.registry.discover_all",
+        "nanobot.channels.registry.discover_enabled",
         return_value={"fakeplugin": _FakePlugin},
     ):
         mgr = ChannelManager.__new__(ChannelManager)
@@ -210,7 +246,7 @@ async def test_manager_propagates_groq_transcription_api_base_to_channels():
     )
 
     with patch(
-        "nanobot.channels.registry.discover_all",
+        "nanobot.channels.registry.discover_enabled",
         return_value={"fakeplugin": _FakePlugin},
     ):
         mgr = ChannelManager.__new__(ChannelManager)
@@ -246,7 +282,7 @@ async def test_manager_propagates_openai_transcription_api_base_to_channels():
     )
 
     with patch(
-        "nanobot.channels.registry.discover_all",
+        "nanobot.channels.registry.discover_enabled",
         return_value={"fakeplugin": _FakePlugin},
     ):
         mgr = ChannelManager.__new__(ChannelManager)
@@ -498,10 +534,8 @@ async def test_manager_skips_disabled_plugin():
         providers=SimpleNamespace(groq=SimpleNamespace(api_key="")),
     )
 
-    with patch(
+    ep = _make_entry_point("fakeplugin", _FakePlugin)
-        "nanobot.channels.registry.discover_all",
+    with patch(_EP_TARGET, return_value=[ep]):
-        return_value={"fakeplugin": _FakePlugin},
-    ):
         mgr = ChannelManager.__new__(ChannelManager)
         mgr.config = fake_config
         mgr.bus = MessageBus()

tests/channels/test_feishu_post_content.py

@@ -27,11 +27,10 @@ def test_extract_post_content_supports_post_wrapper_shape() -> None:
         }
     }
 
-    text, image_keys, media_items = _extract_post_content(payload)
+    text, image_keys = _extract_post_content(payload)
 
     assert text == "日报 完成"
     assert image_keys == ["img_1"]
-    assert media_items == []
 
 
 def test_extract_post_content_keeps_direct_shape_behavior() -> None:
@@ -46,29 +45,10 @@ def test_extract_post_content_keeps_direct_shape_behavior() -> None:
         ],
     }
 
-    text, image_keys, media_items = _extract_post_content(payload)
+    text, image_keys = _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/v1"
+        assert providers["skywork"]["default_api_base"] == "https://api.apifree.ai/agent/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
+from nanobot.config.schema import Config, ModelPresetConfig
 from nanobot.cron.types import CronJob, CronPayload
 from nanobot.providers.factory import ProviderSnapshot
 from nanobot.providers.openai_codex_provider import _strip_model_prefix
@@ -226,6 +226,16 @@ 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):
@@ -280,6 +290,175 @@ 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"])
 
@@ -398,6 +577,8 @@ 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
@@ -540,6 +721,23 @@ 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
 
@@ -572,6 +770,7 @@ 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")
 
@@ -611,7 +810,8 @@ def test_make_provider_passes_extra_headers_to_custom_provider():
     )
 
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai:
-        make_provider(config)
+        provider = make_provider(config)
+        asyncio.run(provider._ensure_client())
 
     kwargs = mock_async_openai.call_args.kwargs
     assert kwargs["api_key"] == "test-key"

tests/providers/test_github_copilot_routing.py

@@ -65,6 +65,7 @@ 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,27 +449,28 @@ def test_gemma_routes_to_gemini_provider() -> None:
     assert "gemma" in spec.keywords
 
 
-def test_openrouter_sets_default_attribution_headers() -> None:
+async def test_openrouter_sets_default_attribution_headers() -> None:
     spec = find_by_name("openrouter")
-    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as MockClient:
+    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_client_cls:
-        OpenAICompatProvider(
+        provider = 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 = MockClient.call_args.kwargs["default_headers"]
+    headers = mock_client_cls.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
 
 
-def test_openrouter_user_headers_override_default_attribution() -> None:
+async def test_openrouter_user_headers_override_default_attribution() -> None:
     spec = find_by_name("openrouter")
-    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as MockClient:
+    with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_client_cls:
-        OpenAICompatProvider(
+        provider = OpenAICompatProvider(
             api_key="sk-or-test-key",
             api_base="https://openrouter.ai/api/v1",
             default_model="anthropic/claude-sonnet-4-5",
@@ -480,8 +481,9 @@ def test_openrouter_user_headers_override_default_attribution() -> None:
             },
             spec=spec,
         )
+        await provider._ensure_client()
 
-    headers = MockClient.call_args.kwargs["default_headers"]
+    headers = mock_client_cls.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,17 +85,18 @@ class TestIsLocalEndpoint:
 class TestLocalKeepaliveConfig:
     """Verify that local endpoints get keepalive_expiry=0."""
 
-    def test_local_spec_disables_keepalive(self):
+    async 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
 
-    def test_lan_ip_disables_keepalive(self):
+    async 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 = ""
@@ -103,16 +104,18 @@ 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
 
-    def test_cloud_keeps_default_keepalive(self):
+    async 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,16 +8,18 @@ def _assert_openai_compat_timeout(timeout) -> None:
     assert timeout == 120.0
 
 
-def test_openai_compat_provider_sets_sdk_timeout() -> None:
+async def test_openai_compat_provider_defers_sdk_client_until_first_use() -> None:
     with patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai:
-        OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
+        provider = OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
+        mock_async_openai.assert_not_called()
+        await provider._ensure_client()
 
     kwargs = mock_async_openai.call_args.kwargs
     _assert_openai_compat_timeout(kwargs["timeout"])
     assert kwargs["http_client"] is None
 
 
-def test_openai_compat_provider_sets_timeout_on_local_http_client() -> None:
+async def test_openai_compat_provider_sets_timeout_on_local_http_client() -> None:
     spec = ProviderSpec(
         name="local",
         keywords=(),
@@ -29,11 +31,13 @@ def test_openai_compat_provider_sets_timeout_on_local_http_client() -> None:
     with (
         patch("nanobot.providers.openai_compat_provider.AsyncOpenAI") as mock_async_openai,
         patch(
-            "nanobot.providers.openai_compat_provider.httpx.AsyncClient",
+            "httpx.AsyncClient",
             return_value=sentinel.http_client,
         ) as mock_http_client,
     ):
-        OpenAICompatProvider(spec=spec)
+        provider = OpenAICompatProvider(spec=spec)
+        mock_async_openai.assert_not_called()
+        await provider._ensure_client()
 
     client_kwargs = mock_http_client.call_args.kwargs
     _assert_openai_compat_timeout(client_kwargs["timeout"])
@@ -44,10 +48,11 @@ def test_openai_compat_provider_sets_timeout_on_local_http_client() -> None:
     assert openai_kwargs["http_client"] is sentinel.http_client
 
 
-def test_openai_compat_provider_timeout_can_be_overridden_by_env(monkeypatch) -> None:
+async 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:
-        OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
+        provider = OpenAICompatProvider(api_key="test-key", api_base="https://example.com/v1")
+        await provider._ensure_client()
 
     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]'."""
+    """When _meta is absent, fallback placeholder is '[image omitted]'."""
     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]" in (b.get("text") or "") for b in content)
+            assert any("[image omitted]" in (b.get("text") or "") for b in content)
 
 
 @pytest.mark.asyncio

tests/providers/test_provider_sdk_retry_defaults.py

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

tests/providers/test_providers_init.py

@@ -12,6 +12,7 @@ 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)
 
@@ -21,6 +22,7 @@ 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__ == [
@@ -30,6 +32,7 @@ def test_importing_providers_package_is_lazy(monkeypatch) -> None:
         "OpenAICompatProvider",
         "OpenAICodexProvider",
         "GitHubCopilotProvider",
+        "XaiOAuthProvider",
         "AzureOpenAIProvider",
         "BedrockProvider",
     ]
@@ -50,3 +53,9 @@ 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/v1"
+    assert skywork.default_api_base == "https://api.apifree.ai/agent/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/v1"
+    assert config.get_api_base("skywork-ai/skyclaw-v1") == "https://api.apifree.ai/agent/v1"
 
 
 def test_skywork_preserves_model_id_and_uses_chat_completion_max_tokens() -> None:

tests/providers/test_xai_oauth_auth.py

@@ -0,0 +1,146 @@
+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

@@ -0,0 +1,141 @@
+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,47 +29,3 @@ 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,6 +5,7 @@ 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
 
@@ -108,6 +109,9 @@ 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:
 
@@ -124,6 +128,9 @@ 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"}