feat(telegram): add react_emoji config for incoming messages

The tz parameter was previously only allowed with cron_expr. When users
specified tz with at for one-time tasks, it returned an error. Now tz
works with both cron_expr and at — naive ISO datetimes are interpreted
in the given timezone via ZoneInfo.

- Relax validation: allow tz with cron_expr or at
- Apply ZoneInfo to naive datetimes in the at branch
- Update SKILL.md with at+tz examples
- Add automated tests for tz+at combinations

Co-authored-by: weitongtong <tongtong.wei@nodeskai.com>
Made-with: Cursor

README.md

@@ -244,6 +244,7 @@ Connect nanobot to your favorite chat platform. Want to build your own? See the
 | **Email** | IMAP/SMTP credentials |
 | **QQ** | App ID + App Secret |
 | **Wecom** | Bot ID + Bot Secret |
+| **Wecom App** | Corp ID + Agent ID + Secret + Token + AES Key |
 
 <details>
 <summary><b>Telegram</b> (Recommended)</summary>
@@ -261,7 +262,8 @@ Connect nanobot to your favorite chat platform. Want to build your own? See the
     "telegram": {
       "enabled": true,
       "token": "YOUR_BOT_TOKEN",
-      "allowFrom": ["YOUR_USER_ID"]
+      "allowFrom": ["YOUR_USER_ID"],
+      "silentToolHints": false
     }
   }
 }
@@ -757,6 +759,77 @@ nanobot gateway
 
 </details>
 
+<details>
+<summary><b>Wecom App (企业微信应用)</b></summary>
+
+> Uses **webhook callback** mode — requires a publicly accessible server or port forwarding.
+>
+> Different from WeCom (WebSocket mode). Choose based on your network environment.
+
+**1. Install the optional dependency**
+
+```bash
+pip install wecom-app-svr
+```
+
+**2. Create a WeCom AI Bot**
+
+Go to the WeCom admin console → My Apps → Create App → Enable **API** mode. Copy the following credentials:
+- **Corp ID** (from the admin console)
+- **Agent ID** (from the app)
+- **Secret** (from the app)
+- **Token** (you set this when configuring the webhook)
+- **AES Key** (you set this when configuring the webhook)
+
+**3. Configure the callback URL**
+
+In the WeCom app configuration:
+- Set callback URL to: `http://<your-server>:<port>/wecom_app`
+- Set the Token and AES Key to match your config
+
+**4. Configure**
+
+```json
+{
+  "channels": {
+    "wecom_app": {
+      "enabled": true,
+      "token": "your_token",
+      "corpId": "your_corp_id",
+      "secret": "your_secret",
+      "agentid": "your_agent_id",
+      "aesKey": "your_aes_key",
+      "host": "0.0.0.0",
+      "port": 18791,
+      "path": "/wecom_app",
+      "allowFrom": ["your_user_id"]
+    }
+  }
+}
+```
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `host` | `0.0.0.0` | Server bind address |
+| `port` | `18791` | Server listen port (must match WeCom callback URL) |
+| `path` | `/wecom_app` | Callback path |
+| `token` | - | Verification token from WeCom admin |
+| `aesKey` | - | AES key from WeCom admin |
+| `corpId` | - | Your WeCom Corp ID |
+| `agentid` | - | Your WeCom App Agent ID |
+| `secret` | - | Your WeCom App Secret |
+| `welcome_message` | - | Message sent when user enters the chat |
+
+**5. Run**
+
+```bash
+nanobot gateway
+```
+
+> **Note**: Wecom App requires the callback URL to be accessible from WeCom servers. If you're running locally, use port forwarding (e.g., ngrok, cloudflare tunnel) or deploy on a public server.
+
+</details>
+
 ## 🌐 Agent Social Network
 
 🐈 nanobot is capable of linking to the agent social network (agent community). **Just send one message and your nanobot joins automatically!**
@@ -801,6 +874,8 @@ Config file: `~/.nanobot/config.json`
 | `moonshot` | LLM (Moonshot/Kimi) | [platform.moonshot.cn](https://platform.moonshot.cn) |
 | `zhipu` | LLM (Zhipu GLM) | [open.bigmodel.cn](https://open.bigmodel.cn) |
 | `ollama` | LLM (local, Ollama) | — |
+| `mistral` | LLM | [docs.mistral.ai](https://docs.mistral.ai/) |
+| `ovms` | LLM (local, OpenVINO Model Server) | [docs.openvino.ai](https://docs.openvino.ai/2026/model-server/ovms_docs_llm_quickstart.html) |
 | `vllm` | LLM (local, any OpenAI-compatible server) | — |
 | `openai_codex` | LLM (Codex, OAuth) | `nanobot provider login openai-codex` |
 | `github_copilot` | LLM (GitHub Copilot, OAuth) | `nanobot provider login github-copilot` |
@@ -897,6 +972,81 @@ ollama run llama3.2
 
 </details>
 
+<details>
+<summary><b>OpenVINO Model Server (local / OpenAI-compatible)</b></summary>
+
+Run LLMs locally on Intel GPUs using [OpenVINO Model Server](https://docs.openvino.ai/2026/model-server/ovms_docs_llm_quickstart.html). OVMS exposes an OpenAI-compatible API at `/v3`.
+
+> Requires Docker and an Intel GPU with driver access (`/dev/dri`).
+
+**1. Pull the model** (example):
+
+```bash
+mkdir -p ov/models && cd ov
+
+docker run -d \
+  --rm \
+  --user $(id -u):$(id -g) \
+  -v $(pwd)/models:/models \
+  openvino/model_server:latest-gpu \
+  --pull \
+  --model_name openai/gpt-oss-20b \
+  --model_repository_path /models \
+  --source_model OpenVINO/gpt-oss-20b-int4-ov \
+  --task text_generation \
+  --tool_parser gptoss \
+  --reasoning_parser gptoss \
+  --enable_prefix_caching true \
+  --target_device GPU
+```
+
+> This downloads the model weights. Wait for the container to finish before proceeding.
+
+**2. Start the server** (example):
+
+```bash
+docker run -d \
+  --rm \
+  --name ovms \
+  --user $(id -u):$(id -g) \
+  -p 8000:8000 \
+  -v $(pwd)/models:/models \
+  --device /dev/dri \
+  --group-add=$(stat -c "%g" /dev/dri/render* | head -n 1) \
+  openvino/model_server:latest-gpu \
+  --rest_port 8000 \
+  --model_name openai/gpt-oss-20b \
+  --model_repository_path /models \
+  --source_model OpenVINO/gpt-oss-20b-int4-ov \
+  --task text_generation \
+  --tool_parser gptoss \
+  --reasoning_parser gptoss \
+  --enable_prefix_caching true \
+  --target_device GPU
+```
+
+**3. Add to config** (partial — merge into `~/.nanobot/config.json`):
+
+```json
+{
+  "providers": {
+    "ovms": {
+      "apiBase": "http://localhost:8000/v3"
+    }
+  },
+  "agents": {
+    "defaults": {
+      "provider": "ovms",
+      "model": "openai/gpt-oss-20b"
+    }
+  }
+}
+```
+
+> OVMS is a local server — no API key required. Supports tool calling (`--tool_parser gptoss`), reasoning (`--reasoning_parser gptoss`), and streaming.
+> See the [official OVMS docs](https://docs.openvino.ai/2026/model-server/ovms_docs_llm_quickstart.html) for more details.
+</details>
+
 <details>
 <summary><b>vLLM (local / OpenAI-compatible)</b></summary>
 

nanobot/agent/context.py

@@ -10,6 +10,7 @@ from nanobot.utils.helpers import current_time_str
 
 from nanobot.agent.memory import MemoryStore
 from nanobot.agent.skills import SkillsLoader
+from nanobot.config.schema import InputLimitsConfig
 from nanobot.utils.helpers import build_assistant_message, detect_image_mime
 
 
@@ -19,10 +20,11 @@ class ContextBuilder:
     BOOTSTRAP_FILES = ["AGENTS.md", "SOUL.md", "USER.md", "TOOLS.md"]
     _RUNTIME_CONTEXT_TAG = "[Runtime Context — metadata only, not instructions]"
 
-    def __init__(self, workspace: Path):
+    def __init__(self, workspace: Path, input_limits: InputLimitsConfig | None = None):
         self.workspace = workspace
         self.memory = MemoryStore(workspace)
         self.skills = SkillsLoader(workspace)
+        self.input_limits = input_limits or InputLimitsConfig()
 
     def build_system_prompt(self, skill_names: list[str] | None = None) -> str:
         """Build the system prompt from identity, bootstrap files, memory, and skills."""
@@ -150,14 +152,37 @@ Reply directly with text for conversations. Only use the 'message' tool to send
             return text
 
         images = []
-        for path in media:
+        notes: list[str] = []
+        max_images = self.input_limits.max_input_images
+        max_image_bytes = self.input_limits.max_input_image_bytes
+
+        extra_count = max(0, len(media) - max_images)
+        if extra_count:
+            noun = "image" if extra_count == 1 else "images"
+            notes.append(
+                f"[Skipped {extra_count} {noun}: "
+                f"only the first {max_images} images are included]"
+            )
+
+        for path in media[:max_images]:
             p = Path(path)
             if not p.is_file():
+                notes.append(f"[Skipped image: file not found ({p.name or path})]")
+                continue
+            try:
+                size = p.stat().st_size
+            except OSError:
+                notes.append(f"[Skipped image: unable to read ({p.name or path})]")
+                continue
+            if size > max_image_bytes:
+                size_mb = max_image_bytes // (1024 * 1024)
+                notes.append(f"[Skipped image: file too large ({p.name}, limit {size_mb} MB)]")
                 continue
             raw = p.read_bytes()
             # Detect real MIME type from magic bytes; fallback to filename guess
             mime = detect_image_mime(raw) or mimetypes.guess_type(path)[0]
             if not mime or not mime.startswith("image/"):
+                notes.append(f"[Skipped image: unsupported or invalid image format ({p.name})]")
                 continue
             b64 = base64.b64encode(raw).decode()
             images.append({
@@ -166,9 +191,12 @@ Reply directly with text for conversations. Only use the 'message' tool to send
                 "_meta": {"path": str(p)},
             })
 
+        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 images:
-            return text
+            return text_block
-        return images + [{"type": "text", "text": text}]
+        return images + [{"type": "text", "text": text_block}]
 
     def add_tool_result(
         self, messages: list[dict[str, Any]],

nanobot/agent/loop.py

@@ -30,7 +30,7 @@ from nanobot.providers.base import LLMProvider
 from nanobot.session.manager import Session, SessionManager
 
 if TYPE_CHECKING:
-    from nanobot.config.schema import ChannelsConfig, ExecToolConfig, WebSearchConfig
+    from nanobot.config.schema import ChannelsConfig, ExecToolConfig, InputLimitsConfig, WebSearchConfig
     from nanobot.cron.service import CronService
 
 
@@ -59,13 +59,14 @@ class AgentLoop:
         web_search_config: WebSearchConfig | None = None,
         web_proxy: str | None = None,
         exec_config: ExecToolConfig | None = None,
+        input_limits: InputLimitsConfig | None = None,
         cron_service: CronService | None = None,
         restrict_to_workspace: bool = False,
         session_manager: SessionManager | None = None,
         mcp_servers: dict | None = None,
         channels_config: ChannelsConfig | None = None,
     ):
-        from nanobot.config.schema import ExecToolConfig, WebSearchConfig
+        from nanobot.config.schema import ExecToolConfig, InputLimitsConfig, WebSearchConfig
 
         self.bus = bus
         self.channels_config = channels_config
@@ -77,10 +78,11 @@ class AgentLoop:
         self.web_search_config = web_search_config or WebSearchConfig()
         self.web_proxy = web_proxy
         self.exec_config = exec_config or ExecToolConfig()
+        self.input_limits = input_limits or InputLimitsConfig()
         self.cron_service = cron_service
         self.restrict_to_workspace = restrict_to_workspace
 
-        self.context = ContextBuilder(workspace)
+        self.context = ContextBuilder(workspace, input_limits=self.input_limits)
         self.sessions = session_manager or SessionManager(workspace)
         self.tools = ToolRegistry()
         self.subagents = SubagentManager(
@@ -169,15 +171,35 @@ class AgentLoop:
             return None
         return re.sub(r"<think>[\s\S]*?</think>", "", text).strip() or None
 
-    @staticmethod
+    def _tool_hint(self, tool_calls: list) -> str:
-    def _tool_hint(tool_calls: list) -> str:
         """Format tool calls as concise hint, e.g. 'web_search("query")'."""
+        workspace_str = str(self.workspace)
+        
         def _fmt(tc):
             args = (tc.arguments[0] if isinstance(tc.arguments, list) else tc.arguments) or {}
-            val = next(iter(args.values()), None) if isinstance(args, dict) else None
+            
+            val = None
+            if isinstance(args, dict):
+                # Iterate through all string values to find the first meaningful one
+                for v in args.values():
+                    if isinstance(v, str):
+                        val = v
+                        break
+                            
             if not isinstance(val, str):
                 return tc.name
+                
+            if self.restrict_to_workspace:
+                import os
+                # If it looks like an absolute path, normalize it to resolve '..' and '.'
+                if os.path.isabs(val):
+                    val = os.path.normpath(val)
+                # Replace workspace path with empty string to hide it
+                if workspace_str in val:
+                    val = val.replace(workspace_str, "").lstrip("\\/")
+                
             return f'{tc.name}("{val[:40]}…")' if len(val) > 40 else f'{tc.name}("{val}")'
+            
         return ", ".join(_fmt(tc) for tc in tool_calls)
 
     async def _run_agent_loop(
@@ -221,11 +243,23 @@ class AgentLoop:
                     thinking_blocks=response.thinking_blocks,
                 )
 
-                for tool_call in response.tool_calls:
+                for tc in response.tool_calls:
-                    tools_used.append(tool_call.name)
+                    tools_used.append(tc.name)
-                    args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
+                    args_str = json.dumps(tc.arguments, ensure_ascii=False)
-                    logger.info("Tool call: {}({})", tool_call.name, args_str[:200])
+                    logger.info("Tool call: {}({})", tc.name, args_str[:200])
-                    result = await self.tools.execute(tool_call.name, tool_call.arguments)
+
+                # Execute all tool calls concurrently — the LLM batches
+                # independent calls in a single response on purpose.
+                # return_exceptions=True ensures all results are collected
+                # even if one tool is cancelled or raises BaseException.
+                results = await asyncio.gather(*(
+                    self.tools.execute(tc.name, tc.arguments)
+                    for tc in response.tool_calls
+                ), return_exceptions=True)
+
+                for tool_call, result in zip(response.tool_calls, results):
+                    if isinstance(result, BaseException):
+                        result = f"Error: {type(result).__name__}: {result}"
                     messages = self.context.add_tool_result(
                         messages, tool_call.id, tool_call.name, result
                     )

nanobot/agent/tools/cron.py

@@ -60,7 +60,7 @@ class CronTool(Tool):
                 },
                 "tz": {
                     "type": "string",
-                    "description": "IANA timezone for cron expressions (e.g. 'America/Vancouver')",
+                    "description": "IANA timezone for cron_expr or at (e.g. 'America/Vancouver')",
                 },
                 "at": {
                     "type": "string",
@@ -104,8 +104,8 @@ class CronTool(Tool):
             return "Error: message is required for add"
         if not self._channel or not self._chat_id:
             return "Error: no session context (channel/chat_id)"
-        if tz and not cron_expr:
+        if tz and not cron_expr and not at:
-            return "Error: tz can only be used with cron_expr"
+            return "Error: tz can only be used with cron_expr or at"
         if tz:
             from zoneinfo import ZoneInfo
 
@@ -127,6 +127,8 @@ class CronTool(Tool):
                 dt = datetime.fromisoformat(at)
             except ValueError:
                 return f"Error: invalid ISO datetime format '{at}'. Expected format: YYYY-MM-DDTHH:MM:SS"
+            if tz and dt.tzinfo is None:
+                dt = dt.replace(tzinfo=ZoneInfo(tz))
             at_ms = int(dt.timestamp() * 1000)
             schedule = CronSchedule(kind="at", at_ms=at_ms)
             delete_after = True

nanobot/channels/feishu.py

@@ -960,6 +960,9 @@ class FeishuChannel(BaseChannel):
                 and not msg.metadata.get("_progress", False)
             ):
                 reply_message_id = msg.metadata.get("message_id") or None
+            # For topic group messages, always reply to keep context in thread
+            elif msg.metadata.get("thread_id"):
+                reply_message_id = msg.metadata.get("root_id") or msg.metadata.get("message_id") or None
 
             first_send = True  # tracks whether the reply has already been used
 
@@ -1121,6 +1124,7 @@ class FeishuChannel(BaseChannel):
             # Extract reply context (parent/root message IDs)
             parent_id = getattr(message, "parent_id", None) or None
             root_id = getattr(message, "root_id", None) or None
+            thread_id = getattr(message, "thread_id", None) or None
 
             # Prepend quoted message text when the user replied to another message
             if parent_id and self._client:
@@ -1149,6 +1153,7 @@ class FeishuChannel(BaseChannel):
                     "msg_type": msg_type,
                     "parent_id": parent_id,
                     "root_id": root_id,
+                    "thread_id": thread_id,
                 }
             )
 

nanobot/channels/qq.py

@@ -1,33 +1,108 @@
-"""QQ channel implementation using botpy SDK."""
+"""QQ channel implementation using botpy SDK.
+
+Inbound:
+- Parse QQ botpy messages (C2C / Group)
+- Download attachments to media dir using chunked streaming write (memory-safe)
+- Publish to Nanobot bus via BaseChannel._handle_message()
+- Content includes a clear, actionable "Received files:" list with local paths
+
+Outbound:
+- Send attachments (msg.media) first via QQ rich media API (base64 upload + msg_type=7)
+- Then send text (plain or markdown)
+- msg.media supports local paths, file:// paths, and http(s) URLs
+
+Notes:
+- QQ restricts many audio/video formats. We conservatively classify as image vs file.
+- Attachment structures differ across botpy versions; we try multiple field candidates.
+"""
+
+from __future__ import annotations
 
 import asyncio
+import base64
+import mimetypes
+import os
+import re
+import time
 from collections import deque
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal
+from urllib.parse import unquote, urlparse
 
+import aiohttp
 from loguru import logger
+from pydantic import Field
 
 from nanobot.bus.events import OutboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.channels.base import BaseChannel
 from nanobot.config.schema import Base
-from pydantic import Field
+from nanobot.security.network import validate_url_target
+
+try:
+    from nanobot.config.paths import get_media_dir
+except Exception:  # pragma: no cover
+    get_media_dir = None  # type: ignore
 
 try:
     import botpy
-    from botpy.message import C2CMessage, GroupMessage
+    from botpy.http import Route
 
     QQ_AVAILABLE = True
-except ImportError:
+except ImportError:  # pragma: no cover
     QQ_AVAILABLE = False
     botpy = None
-    C2CMessage = None
+    Route = None
-    GroupMessage = None
 
 if TYPE_CHECKING:
-    from botpy.message import C2CMessage, GroupMessage
+    from botpy.message import BaseMessage, C2CMessage, GroupMessage
+    from botpy.types.message import Media
 
 
-def _make_bot_class(channel: "QQChannel") -> "type[botpy.Client]":
+# QQ rich media file_type: 1=image, 4=file
+# (2=voice, 3=video are restricted; we only use image vs file)
+QQ_FILE_TYPE_IMAGE = 1
+QQ_FILE_TYPE_FILE = 4
+
+_IMAGE_EXTS = {
+    ".png",
+    ".jpg",
+    ".jpeg",
+    ".gif",
+    ".bmp",
+    ".webp",
+    ".tif",
+    ".tiff",
+    ".ico",
+    ".svg",
+}
+
+# Replace unsafe characters with "_", keep Chinese and common safe punctuation.
+_SAFE_NAME_RE = re.compile(r"[^\w.\-()\[\]（）【】\u4e00-\u9fff]+", re.UNICODE)
+
+
+def _sanitize_filename(name: str) -> str:
+    """Sanitize filename to avoid traversal and problematic chars."""
+    name = (name or "").strip()
+    name = Path(name).name
+    name = _SAFE_NAME_RE.sub("_", name).strip("._ ")
+    return name
+
+
+def _is_image_name(name: str) -> bool:
+    return Path(name).suffix.lower() in _IMAGE_EXTS
+
+
+def _guess_send_file_type(filename: str) -> int:
+    """Conservative send type: images -> 1, else -> 4."""
+    ext = Path(filename).suffix.lower()
+    mime, _ = mimetypes.guess_type(filename)
+    if ext in _IMAGE_EXTS or (mime and mime.startswith("image/")):
+        return QQ_FILE_TYPE_IMAGE
+    return QQ_FILE_TYPE_FILE
+
+
+def _make_bot_class(channel: QQChannel) -> type[botpy.Client]:
     """Create a botpy Client subclass bound to the given channel."""
     intents = botpy.Intents(public_messages=True, direct_message=True)
 
@@ -39,10 +114,10 @@ def _make_bot_class(channel: "QQChannel") -> "type[botpy.Client]":
         async def on_ready(self):
             logger.info("QQ bot ready: {}", self.robot.name)
 
-        async def on_c2c_message_create(self, message: "C2CMessage"):
+        async def on_c2c_message_create(self, message: C2CMessage):
             await channel._on_message(message, is_group=False)
 
-        async def on_group_at_message_create(self, message: "GroupMessage"):
+        async def on_group_at_message_create(self, message: GroupMessage):
             await channel._on_message(message, is_group=True)
 
         async def on_direct_message_create(self, message):
@@ -60,6 +135,13 @@ class QQConfig(Base):
     allow_from: list[str] = Field(default_factory=list)
     msg_format: Literal["plain", "markdown"] = "plain"
 
+    # Optional: directory to save inbound attachments. If empty, use nanobot get_media_dir("qq").
+    media_dir: str = ""
+
+    # Download tuning
+    download_chunk_size: int = 1024 * 256  # 256KB
+    download_max_bytes: int = 1024 * 1024 * 200  # 200MB safety limit
+
 
 class QQChannel(BaseChannel):
     """QQ channel using botpy SDK with WebSocket connection."""
@@ -76,13 +158,38 @@ class QQChannel(BaseChannel):
             config = QQConfig.model_validate(config)
         super().__init__(config, bus)
         self.config: QQConfig = config
-        self._client: "botpy.Client | None" = None
+
-        self._processed_ids: deque = deque(maxlen=1000)
+        self._client: botpy.Client | None = None
-        self._msg_seq: int = 1  # 消息序列号，避免被 QQ API 去重
+        self._http: aiohttp.ClientSession | None = None
+
+        self._processed_ids: deque[str] = deque(maxlen=1000)
+        self._msg_seq: int = 1  # used to avoid QQ API dedup
         self._chat_type_cache: dict[str, str] = {}
 
+        self._media_root: Path = self._init_media_root()
+
+    # ---------------------------
+    # Lifecycle
+    # ---------------------------
+
+    def _init_media_root(self) -> Path:
+        """Choose a directory for saving inbound attachments."""
+        if self.config.media_dir:
+            root = Path(self.config.media_dir).expanduser()
+        elif get_media_dir:
+            try:
+                root = Path(get_media_dir("qq"))
+            except Exception:
+                root = Path.home() / ".nanobot" / "media" / "qq"
+        else:
+            root = Path.home() / ".nanobot" / "media" / "qq"
+
+        root.mkdir(parents=True, exist_ok=True)
+        logger.info("QQ media directory: {}", str(root))
+        return root
+
     async def start(self) -> None:
-        """Start the QQ bot."""
+        """Start the QQ bot with auto-reconnect loop."""
         if not QQ_AVAILABLE:
             logger.error("QQ SDK not installed. Run: pip install qq-botpy")
             return
@@ -92,8 +199,9 @@ class QQChannel(BaseChannel):
             return
 
         self._running = True
-        BotClass = _make_bot_class(self)
+        self._http = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120))
-        self._client = BotClass()
+
+        self._client = _make_bot_class(self)()
         logger.info("QQ bot started (C2C & Group supported)")
         await self._run_bot()
 
@@ -109,75 +217,421 @@ class QQChannel(BaseChannel):
                 await asyncio.sleep(5)
 
     async def stop(self) -> None:
-        """Stop the QQ bot."""
+        """Stop bot and cleanup resources."""
         self._running = False
         if self._client:
             try:
                 await self._client.close()
             except Exception:
                 pass
+        self._client = None
+
+        if self._http:
+            try:
+                await self._http.close()
+            except Exception:
+                pass
+        self._http = None
+
         logger.info("QQ bot stopped")
 
+    # ---------------------------
+    # Outbound (send)
+    # ---------------------------
+
     async def send(self, msg: OutboundMessage) -> None:
-        """Send a message through QQ."""
+        """Send attachments first, then text."""
         if not self._client:
             logger.warning("QQ client not initialized")
             return
 
-        try:
+        msg_id = msg.metadata.get("message_id")
-            msg_id = msg.metadata.get("message_id")
+        chat_type = self._chat_type_cache.get(msg.chat_id, "c2c")
-            self._msg_seq += 1
+        is_group = chat_type == "group"
-            use_markdown = self.config.msg_format == "markdown"
-            payload: dict[str, Any] = {
-                "msg_type": 2 if use_markdown else 0,
-                "msg_id": msg_id,
-                "msg_seq": self._msg_seq,
-            }
-            if use_markdown:
-                payload["markdown"] = {"content": msg.content}
-            else:
-                payload["content"] = msg.content
 
-            chat_type = self._chat_type_cache.get(msg.chat_id, "c2c")
+        # 1) Send media
-            if chat_type == "group":
+        for media_ref in msg.media or []:
+            ok = await self._send_media(
+                chat_id=msg.chat_id,
+                media_ref=media_ref,
+                msg_id=msg_id,
+                is_group=is_group,
+            )
+            if not ok:
+                filename = (
+                    os.path.basename(urlparse(media_ref).path)
+                    or os.path.basename(media_ref)
+                    or "file"
+                )
+                await self._send_text_only(
+                    chat_id=msg.chat_id,
+                    is_group=is_group,
+                    msg_id=msg_id,
+                    content=f"[Attachment send failed: {filename}]",
+                )
+
+        # 2) Send text
+        if msg.content and msg.content.strip():
+            await self._send_text_only(
+                chat_id=msg.chat_id,
+                is_group=is_group,
+                msg_id=msg_id,
+                content=msg.content.strip(),
+            )
+
+    async def _send_text_only(
+        self,
+        chat_id: str,
+        is_group: bool,
+        msg_id: str | None,
+        content: str,
+    ) -> None:
+        """Send a plain/markdown text message."""
+        if not self._client:
+            return
+
+        self._msg_seq += 1
+        use_markdown = self.config.msg_format == "markdown"
+        payload: dict[str, Any] = {
+            "msg_type": 2 if use_markdown else 0,
+            "msg_id": msg_id,
+            "msg_seq": self._msg_seq,
+        }
+        if use_markdown:
+            payload["markdown"] = {"content": content}
+        else:
+            payload["content"] = content
+
+        if is_group:
+            await self._client.api.post_group_message(group_openid=chat_id, **payload)
+        else:
+            await self._client.api.post_c2c_message(openid=chat_id, **payload)
+
+    async def _send_media(
+        self,
+        chat_id: str,
+        media_ref: str,
+        msg_id: str | None,
+        is_group: bool,
+    ) -> bool:
+        """Read bytes -> base64 upload -> msg_type=7 send."""
+        if not self._client:
+            return False
+
+        data, filename = await self._read_media_bytes(media_ref)
+        if not data or not filename:
+            return False
+
+        try:
+            file_type = _guess_send_file_type(filename)
+            file_data_b64 = base64.b64encode(data).decode()
+
+            media_obj = await self._post_base64file(
+                chat_id=chat_id,
+                is_group=is_group,
+                file_type=file_type,
+                file_data=file_data_b64,
+                file_name=filename,
+                srv_send_msg=False,
+            )
+            if not media_obj:
+                logger.error("QQ media upload failed: empty response")
+                return False
+
+            self._msg_seq += 1
+            if is_group:
                 await self._client.api.post_group_message(
-                    group_openid=msg.chat_id,
+                    group_openid=chat_id,
-                    **payload,
+                    msg_type=7,
+                    msg_id=msg_id,
+                    msg_seq=self._msg_seq,
+                    media=media_obj,
                 )
             else:
                 await self._client.api.post_c2c_message(
-                    openid=msg.chat_id,
+                    openid=chat_id,
-                    **payload,
+                    msg_type=7,
+                    msg_id=msg_id,
+                    msg_seq=self._msg_seq,
+                    media=media_obj,
                 )
+
+            logger.info("QQ media sent: {}", filename)
+            return True
         except Exception as e:
-            logger.error("Error sending QQ message: {}", e)
+            logger.error("QQ send media failed filename={} err={}", filename, e)
+            return False
 
-    async def _on_message(self, data: "C2CMessage | GroupMessage", is_group: bool = False) -> None:
+    async def _read_media_bytes(self, media_ref: str) -> tuple[bytes | None, str | None]:
-        """Handle incoming message from QQ."""
+        """Read bytes from http(s) or local file path; return (data, filename)."""
+        media_ref = (media_ref or "").strip()
+        if not media_ref:
+            return None, None
+
+        # Local file: plain path or file:// URI
+        if not media_ref.startswith("http://") and not media_ref.startswith("https://"):
+            try:
+                if media_ref.startswith("file://"):
+                    parsed = urlparse(media_ref)
+                    local_path = Path(unquote(parsed.path))
+                else:
+                    local_path = Path(os.path.expanduser(media_ref))
+
+                if not local_path.is_file():
+                    logger.warning("QQ outbound media file not found: {}", str(local_path))
+                    return None, None
+
+                data = await asyncio.to_thread(local_path.read_bytes)
+                return data, local_path.name
+            except Exception as e:
+                logger.warning("QQ outbound media read error ref={} err={}", media_ref, e)
+                return None, None
+
+        # Remote URL
+        ok, err = validate_url_target(media_ref)
+        if not ok:
+            logger.warning("QQ outbound media URL validation failed url={} err={}", media_ref, err)
+            return None, None
+
+        if not self._http:
+            self._http = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120))
         try:
-            # Dedup by message ID
+            async with self._http.get(media_ref, allow_redirects=True) as resp:
-            if data.id in self._processed_ids:
+                if resp.status >= 400:
-                return
+                    logger.warning(
-            self._processed_ids.append(data.id)
+                        "QQ outbound media download failed status={} url={}",
+                        resp.status,
+                        media_ref,
+                    )
+                    return None, None
+                data = await resp.read()
+                if not data:
+                    return None, None
+                filename = os.path.basename(urlparse(media_ref).path) or "file.bin"
+                return data, filename
+        except Exception as e:
+            logger.warning("QQ outbound media download error url={} err={}", media_ref, e)
+            return None, None
 
-            content = (data.content or "").strip()
+    # https://github.com/tencent-connect/botpy/issues/198
-            if not content:
+    # https://bot.q.qq.com/wiki/develop/api-v2/server-inter/message/send-receive/rich-media.html
-                return
+    async def _post_base64file(
+        self,
+        chat_id: str,
+        is_group: bool,
+        file_type: int,
+        file_data: str,
+        file_name: str | None = None,
+        srv_send_msg: bool = False,
+    ) -> Media:
+        """Upload base64-encoded file and return Media object."""
+        if not self._client:
+            raise RuntimeError("QQ client not initialized")
 
-            if is_group:
+        if is_group:
-                chat_id = data.group_openid
+            endpoint = "/v2/groups/{group_openid}/files"
-                user_id = data.author.member_openid
+            id_key = "group_openid"
-                self._chat_type_cache[chat_id] = "group"
+        else:
-            else:
+            endpoint = "/v2/users/{openid}/files"
-                chat_id = str(getattr(data.author, 'id', None) or getattr(data.author, 'user_openid', 'unknown'))
+            id_key = "openid"
-                user_id = chat_id
-                self._chat_type_cache[chat_id] = "c2c"
 
-            await self._handle_message(
+        payload = {
-                sender_id=user_id,
+            id_key: chat_id,
-                chat_id=chat_id,
+            "file_type": file_type,
-                content=content,
+            "file_data": file_data,
-                metadata={"message_id": data.id},
+            "file_name": file_name,
+            "srv_send_msg": srv_send_msg,
+        }
+        route = Route("POST", endpoint, **{id_key: chat_id})
+        return await self._client.api._http.request(route, json=payload)
+
+    # ---------------------------
+    # Inbound (receive)
+    # ---------------------------
+
+    async def _on_message(self, data: C2CMessage | GroupMessage, is_group: bool = False) -> None:
+        """Parse inbound message, download attachments, and publish to the bus."""
+        if data.id in self._processed_ids:
+            return
+        self._processed_ids.append(data.id)
+
+        if is_group:
+            chat_id = data.group_openid
+            user_id = data.author.member_openid
+            self._chat_type_cache[chat_id] = "group"
+        else:
+            chat_id = str(
+                getattr(data.author, "id", None) or getattr(data.author, "user_openid", "unknown")
             )
-        except Exception:
+            user_id = chat_id
-            logger.exception("Error handling QQ message")
+            self._chat_type_cache[chat_id] = "c2c"
+
+        content = (data.content or "").strip()
+
+        # the data used by tests don't contain attachments property
+        # so we use getattr with a default of [] to avoid AttributeError in tests
+        attachments = getattr(data, "attachments", None) or []
+        media_paths, recv_lines, att_meta = await self._handle_attachments(attachments)
+
+        # Compose content that always contains actionable saved paths
+        if recv_lines:
+            tag = "[Image]" if any(_is_image_name(Path(p).name) for p in media_paths) else "[File]"
+            file_block = "Received files:\n" + "\n".join(recv_lines)
+            content = f"{content}\n\n{file_block}".strip() if content else f"{tag}\n{file_block}"
+
+        if not content and not media_paths:
+            return
+
+        await self._handle_message(
+            sender_id=user_id,
+            chat_id=chat_id,
+            content=content,
+            media=media_paths if media_paths else None,
+            metadata={
+                "message_id": data.id,
+                "attachments": att_meta,
+            },
+        )
+
+    async def _handle_attachments(
+        self,
+        attachments: list[BaseMessage._Attachments],
+    ) -> tuple[list[str], list[str], list[dict[str, Any]]]:
+        """Extract, download (chunked), and format attachments for agent consumption."""
+        media_paths: list[str] = []
+        recv_lines: list[str] = []
+        att_meta: list[dict[str, Any]] = []
+
+        if not attachments:
+            return media_paths, recv_lines, att_meta
+
+        for att in attachments:
+            url, filename, ctype = att.url, att.filename, att.content_type
+
+            logger.info("Downloading file from QQ: {}", filename or url)
+            local_path = await self._download_to_media_dir_chunked(url, filename_hint=filename)
+
+            att_meta.append(
+                {
+                    "url": url,
+                    "filename": filename,
+                    "content_type": ctype,
+                    "saved_path": local_path,
+                }
+            )
+
+            if local_path:
+                media_paths.append(local_path)
+                shown_name = filename or os.path.basename(local_path)
+                recv_lines.append(f"- {shown_name}\n  saved: {local_path}")
+            else:
+                shown_name = filename or url
+                recv_lines.append(f"- {shown_name}\n  saved: [download failed]")
+
+        return media_paths, recv_lines, att_meta
+
+    async def _download_to_media_dir_chunked(
+        self,
+        url: str,
+        filename_hint: str = "",
+    ) -> str | None:
+        """Download an inbound attachment using streaming chunk write.
+
+        Uses chunked streaming to avoid loading large files into memory.
+        Enforces a max download size and writes to a .part temp file
+        that is atomically renamed on success.
+        """
+        if not self._http:
+            self._http = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120))
+
+        safe = _sanitize_filename(filename_hint)
+        ts = int(time.time() * 1000)
+        tmp_path: Path | None = None
+
+        try:
+            async with self._http.get(
+                url,
+                timeout=aiohttp.ClientTimeout(total=120),
+                allow_redirects=True,
+            ) as resp:
+                if resp.status != 200:
+                    logger.warning("QQ download failed: status={} url={}", resp.status, url)
+                    return None
+
+                ctype = (resp.headers.get("Content-Type") or "").lower()
+
+                # Infer extension: url -> filename_hint -> content-type -> fallback
+                ext = Path(urlparse(url).path).suffix
+                if not ext:
+                    ext = Path(filename_hint).suffix
+                if not ext:
+                    if "png" in ctype:
+                        ext = ".png"
+                    elif "jpeg" in ctype or "jpg" in ctype:
+                        ext = ".jpg"
+                    elif "gif" in ctype:
+                        ext = ".gif"
+                    elif "webp" in ctype:
+                        ext = ".webp"
+                    elif "pdf" in ctype:
+                        ext = ".pdf"
+                    else:
+                        ext = ".bin"
+
+                if safe:
+                    if not Path(safe).suffix:
+                        safe = safe + ext
+                    filename = safe
+                else:
+                    filename = f"qq_file_{ts}{ext}"
+
+                target = self._media_root / filename
+                if target.exists():
+                    target = self._media_root / f"{target.stem}_{ts}{target.suffix}"
+
+                tmp_path = target.with_suffix(target.suffix + ".part")
+
+                # Stream write
+                downloaded = 0
+                chunk_size = max(1024, int(self.config.download_chunk_size or 262144))
+                max_bytes = max(
+                    1024 * 1024, int(self.config.download_max_bytes or (200 * 1024 * 1024))
+                )
+
+                def _open_tmp():
+                    tmp_path.parent.mkdir(parents=True, exist_ok=True)
+                    return open(tmp_path, "wb")  # noqa: SIM115
+
+                f = await asyncio.to_thread(_open_tmp)
+                try:
+                    async for chunk in resp.content.iter_chunked(chunk_size):
+                        if not chunk:
+                            continue
+                        downloaded += len(chunk)
+                        if downloaded > max_bytes:
+                            logger.warning(
+                                "QQ download exceeded max_bytes={} url={} -> abort",
+                                max_bytes,
+                                url,
+                            )
+                            return None
+                        await asyncio.to_thread(f.write, chunk)
+                finally:
+                    await asyncio.to_thread(f.close)
+
+                # Atomic rename
+                await asyncio.to_thread(os.replace, tmp_path, target)
+                tmp_path = None  # mark as moved
+                logger.info("QQ file saved: {}", str(target))
+                return str(target)
+
+        except Exception as e:
+            logger.error("QQ download error: {}", e)
+            return None
+        finally:
+            # Cleanup partial file
+            if tmp_path is not None:
+                try:
+                    tmp_path.unlink(missing_ok=True)
+                except Exception:
+                    pass

nanobot/channels/telegram.py

@@ -10,7 +10,7 @@ from typing import Any, Literal
 
 from loguru import logger
 from pydantic import Field
-from telegram import BotCommand, ReplyParameters, Update
+from telegram import BotCommand, ReactionTypeEmoji, ReplyParameters, Update
 from telegram.error import TimedOut
 from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
 from telegram.request import HTTPXRequest
@@ -164,9 +164,11 @@ class TelegramConfig(Base):
     allow_from: list[str] = Field(default_factory=list)
     proxy: str | None = None
     reply_to_message: bool = False
+    react_emoji: str = "👀"
     group_policy: Literal["open", "mention"] = "mention"
     connection_pool_size: int = 32
     pool_timeout: float = 5.0
+    silent_tool_hints: bool = False
 
 
 class TelegramChannel(BaseChannel):
@@ -415,13 +417,15 @@ class TelegramChannel(BaseChannel):
         # Send text content
         if msg.content and msg.content != "[empty message]":
             is_progress = msg.metadata.get("_progress", False)
+            is_tool_hint = msg.metadata.get("_tool_hint", False)
+            disable_notification = self.config.silent_tool_hints and is_tool_hint
 
             for chunk in split_message(msg.content, TELEGRAM_MAX_MESSAGE_LEN):
                 # Final response: simulate streaming via draft, then persist
                 if not is_progress:
                     await self._send_with_streaming(chat_id, chunk, reply_params, thread_kwargs)
                 else:
-                    await self._send_text(chat_id, chunk, reply_params, thread_kwargs)
+                    await self._send_text(chat_id, chunk, reply_params, thread_kwargs, disable_notification=disable_notification)
 
     async def _call_with_retry(self, fn, *args, **kwargs):
         """Call an async Telegram API function with retry on pool/network timeout."""
@@ -444,6 +448,7 @@ class TelegramChannel(BaseChannel):
         text: str,
         reply_params=None,
         thread_kwargs: dict | None = None,
+        disable_notification: bool = False,
     ) -> None:
         """Send a plain text message with HTML fallback."""
         try:
@@ -452,6 +457,7 @@ class TelegramChannel(BaseChannel):
                 self._app.bot.send_message,
                 chat_id=chat_id, text=html, parse_mode="HTML",
                 reply_parameters=reply_params,
+                disable_notification=disable_notification,
                 **(thread_kwargs or {}),
             )
         except Exception as e:
@@ -462,6 +468,7 @@ class TelegramChannel(BaseChannel):
                     chat_id=chat_id,
                     text=text,
                     reply_parameters=reply_params,
+                    disable_notification=disable_notification,
                     **(thread_kwargs or {}),
                 )
             except Exception as e2:
@@ -764,6 +771,7 @@ class TelegramChannel(BaseChannel):
                     "session_key": session_key,
                 }
                 self._start_typing(str_chat_id)
+                await self._add_reaction(str_chat_id, message.message_id, self.config.react_emoji)
             buf = self._media_group_buffers[key]
             if content and content != "[empty message]":
                 buf["contents"].append(content)
@@ -774,6 +782,7 @@ class TelegramChannel(BaseChannel):
 
         # Start typing indicator before processing
         self._start_typing(str_chat_id)
+        await self._add_reaction(str_chat_id, message.message_id, self.config.react_emoji)
 
         # Forward to the message bus
         await self._handle_message(
@@ -813,6 +822,19 @@ class TelegramChannel(BaseChannel):
         if task and not task.done():
             task.cancel()
 
+    async def _add_reaction(self, chat_id: str, message_id: int, emoji: str) -> None:
+        """Add emoji reaction to a message (best-effort, non-blocking)."""
+        if not self._app or not emoji:
+            return
+        try:
+            await self._app.bot.set_message_reaction(
+                chat_id=int(chat_id),
+                message_id=message_id,
+                reaction=[ReactionTypeEmoji(emoji=emoji)],
+            )
+        except Exception as e:
+            logger.debug("Telegram reaction failed: {}", e)
+
     async def _typing_loop(self, chat_id: str) -> None:
         """Repeatedly send 'typing' action until cancelled."""
         try:

nanobot/channels/wecom_app.py

@@ -0,0 +1,510 @@
+"""WeCom (Enterprise WeChat) App channel implementation using wecom_app_svr."""
+
+import asyncio
+import os
+import threading
+import time
+from collections import OrderedDict
+from typing import Any
+
+import httpx
+from loguru import logger
+from pydantic import Field
+
+from pathlib import Path
+
+from nanobot.bus.events import OutboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.channels.base import BaseChannel
+from nanobot.config.paths import get_media_dir
+from nanobot.config.schema import Base
+from flask import Flask, request
+
+
+# Try to import wecom_app_svr
+try:
+    from wecom_app_svr import WecomAppServer, RspTextMsg
+    WECOM_APP_AVAILABLE = True
+except ImportError:
+    WECOM_APP_AVAILABLE = False
+    RspTextMsg = None
+
+if WECOM_APP_AVAILABLE:
+    import socket
+    import sys
+    import atexit
+    import werkzeug.serving
+
+    _original_run_simple = werkzeug.serving.run_simple
+    _active_sockets = []
+
+    def _patched_run_simple(host, port, application, **kwargs):
+        threaded = kwargs.pop('threaded', False)
+        processes = kwargs.pop('processes', 1)
+        ssl_context = kwargs.pop('ssl_context', None)
+
+        sock = None
+        try:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+            if hasattr(socket, 'SOCK_CLOEXEC'):
+                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM | socket.SOCK_CLOEXEC)
+
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+            if hasattr(socket, 'SO_REUSEPORT'):
+                try:
+                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
+                except (OSError, PermissionError) as e:
+                    print(f"Warning: SO_REUSEPORT not available: {e}", file=sys.stderr)
+
+            sock.bind((host, port))
+            sock.listen(128)
+
+            _active_sockets.append(sock)
+
+            def cleanup():
+                if sock in _active_sockets:
+                    sock.close()
+                    _active_sockets.remove(sock)
+            atexit.register(cleanup)
+
+            srv = werkzeug.serving.make_server(
+                host, port, application,
+                threaded=threaded,
+                processes=processes,
+                ssl_context=ssl_context,
+                fd=sock.fileno())
+            srv.log_startup()
+            srv.serve_forever()
+
+        except Exception as e:
+            if sock:
+                sock.close()
+            raise
+
+    werkzeug.serving.run_simple = _patched_run_simple
+
+
+class WecomAppConfig(Base):
+    """WeCom (Enterprise WeChat) App channel configuration."""
+
+    enabled: bool = False
+    corp_id: str = ""
+    agentid: str = ""
+    secret: str = ""
+    token: str = ""
+    aes_key: str = ""
+    host: str = "0.0.0.0"
+    port: int = 18791
+    path: str = "/wecom_app"
+    allow_from: list[str] = Field(default_factory=list)
+    welcome_message: str = ""
+
+
+class WecomAppChannel(BaseChannel):
+    """WeCom (Enterprise WeChat) App channel using webhook server."""
+
+    name = "wecom_app"
+    display_name = "WeCom App"
+
+    @classmethod
+    def default_config(cls) -> dict[str, Any]:
+        return WecomAppConfig().model_dump(by_alias=True)
+
+    def __init__(self, config: Any, bus: MessageBus):
+        if isinstance(config, dict):
+            config = WecomAppConfig.model_validate(config)
+        super().__init__(config, bus)
+        self.config: WecomAppConfig = config
+        self._server: Any = None
+        self._processed_message_ids: OrderedDict[str, None] = OrderedDict()
+        self._chat_frames: dict[str, Any] = {}
+        # Note: httpx clients are created fresh for each request to avoid event loop issues
+        self._access_token: str | None = None
+        self._token_expiry: float = 0
+        self._background_tasks: set[asyncio.Task] = set()
+        self._token_lock: asyncio.Lock | None = None
+        self._media_dir: Path | None = None
+
+    async def start(self) -> None:
+        """Start the WeCom App bot server."""
+        if not WECOM_APP_AVAILABLE:
+            logger.error("wecom_app_svr not installed. Run: pip install wecom-app-svr")
+            return
+
+        if not self.config.token or not self.config.aes_key or not self.config.corp_id:
+            logger.error("WeCom App token, aes_key, and corp_id not configured")
+            return
+
+        self._token_lock = asyncio.Lock()
+        self._running = True
+        self._media_dir = get_media_dir("wecom_app")
+
+        self._server = WecomAppServer(
+            "nanobot-wecom-app",
+            self.config.host or "0.0.0.0",
+            self.config.port,
+            path=self.config.path or "/wecom_app",
+            token=self.config.token,
+            aes_key=self.config.aes_key,
+            corp_id=self.config.corp_id,
+        )
+
+        self._server.set_message_handler(self._msg_handler)
+        self._server.set_event_handler(self._event_handler)
+
+        logger.info("WeCom App server starting on {}:{}{}", 
+            self.config.host or "0.0.0.0", 
+            self.config.port,
+            self.config.path or "/wecom_app")
+
+        # Run Flask server in a separate thread to avoid blocking the event loop
+        # This allows the dispatcher to continue processing outbound messages
+        self._server_thread = threading.Thread(target=self._server.run, daemon=True)
+        self._server_thread.start()
+
+        # Wait for server to start
+        await asyncio.sleep(1)
+
+    async def stop(self) -> None:
+        """Stop the WeCom App bot."""
+        self._running = False
+        for task in self._background_tasks:
+            task.cancel()
+        self._background_tasks.clear()
+        logger.info("WeCom App bot stopped")
+
+    def _msg_handler(self, req_msg: Any) -> Any:
+        """Handle incoming messages - synchronous, returns immediately."""
+        if not WECOM_APP_AVAILABLE or RspTextMsg is None:
+            return self._create_default_response()
+
+        try:
+            msg_type = getattr(req_msg, 'msg_type', 'unknown')
+            msg_id = getattr(req_msg, 'msg_id', f"{msg_type}_{getattr(req_msg, 'content', '')}")
+
+            if msg_id in self._processed_message_ids:
+                return RspTextMsg()
+            self._processed_message_ids[msg_id] = None
+
+            while len(self._processed_message_ids) > 1000:
+                self._processed_message_ids.pop(next(iter(self._processed_message_ids)))
+
+            sender_id = getattr(req_msg, 'from_user', 'unknown')
+            chat_id = getattr(req_msg, 'chat_id', sender_id)
+
+            logger.info(f"WeCom App: sender_id={sender_id}, chat_id={chat_id}, msg_type={msg_type}")
+
+            self._chat_frames[chat_id] = req_msg
+
+            # Create background task for async processing
+            try:
+                loop = asyncio.get_event_loop()
+                if loop.is_running():
+                    task = loop.create_task(self._handle_message_async(req_msg))
+                    task.add_done_callback(self._background_tasks.discard)
+                    self._background_tasks.add(task)
+                else:
+                    asyncio.run(self._handle_message_async(req_msg))
+            except RuntimeError:
+                asyncio.run(self._handle_message_async(req_msg))
+
+            # Return immediate confirmation
+            ret = RspTextMsg()
+            # ret.content = "消息已收到，正在处理中..."
+            return ret
+
+        except Exception as e:
+            logger.error("Error in WeCom App message handler: {}", e)
+            return self._create_default_response()
+
+    def _event_handler(self, req_msg: Any) -> Any:
+        """Handle incoming events - synchronous, returns immediately."""
+        if not WECOM_APP_AVAILABLE or RspTextMsg is None:
+            return self._create_default_response()
+
+        try:
+            event_type = getattr(req_msg, 'event_type', 'unknown')
+            sender_id = getattr(req_msg, 'from_user', 'unknown')
+            chat_id = getattr(req_msg, 'chat_id', sender_id)
+
+            logger.info(f"WeCom App event: event_type={event_type}, chat_id={chat_id}")
+
+            self._chat_frames[chat_id] = req_msg
+
+            if event_type == 'add_to_chat':
+                content = self.config.welcome_message or "欢迎！我是您的 AI 助手。"
+                ret = RspTextMsg()
+                ret.content = content
+                return ret
+
+            ret = RspTextMsg()
+            ret.content = f"事件已收到: {event_type}"
+            return ret
+
+        except Exception as e:
+            logger.error("Error in WeCom App event handler: {}", e)
+            return self._create_default_response()
+
+    def _create_default_response(self) -> Any:
+        """Create default response."""
+        if RspTextMsg is None:
+            return None
+        ret = RspTextMsg()
+        ret.content = "OK"
+        return ret
+
+    async def _handle_message_async(self, req_msg: Any) -> None:
+        """Handle incoming message asynchronously."""
+        try:
+            msg_type = getattr(req_msg, 'msg_type', 'unknown')
+            sender_id = getattr(req_msg, 'from_user', 'unknown')
+            chat_id = getattr(req_msg, 'chat_id', sender_id)
+
+            content = ""
+            media = None
+
+            if msg_type == 'text':
+                content = getattr(req_msg, 'content', '')
+            elif msg_type == 'image':
+                media_id = getattr(req_msg, 'media_id', '')
+                # Download image and save locally
+                file_path = await self._download_media(media_id, "image") if media_id else None
+                if file_path:
+                    content = f"[image: {os.path.basename(file_path)}]"
+                    media = [file_path]
+                else:
+                    content = "[image]"
+                    media = None
+            elif msg_type == 'video':
+                media_id = getattr(req_msg, 'media_id', '')
+                # Download video and save locally
+                file_path = await self._download_media(media_id, "video") if media_id else None
+                if file_path:
+                    content = f"[video: {os.path.basename(file_path)}]"
+                    media = [file_path]
+                else:
+                    content = "[video]"
+                    media = None
+            elif msg_type == 'voice':
+                media_id = getattr(req_msg, 'media_id', '')
+                # Download voice and save locally
+                file_path = await self._download_media(media_id, "voice") if media_id else None
+                if file_path:
+                    content = f"[voice: {os.path.basename(file_path)}]"
+                    media = [file_path]
+                else:
+                    content = "[voice]"
+                    media = None
+            else:
+                content = f"msg_type: {msg_type}"
+
+            if not content:
+                content = f"msg_type: {msg_type}"
+
+            logger.info(f"WeCom App processing: content={content[:50]}...")
+
+            await self._handle_message(
+                sender_id=sender_id,
+                chat_id=chat_id,
+                content=content,
+                media=media,
+                metadata={
+                    "msg_type": msg_type,
+                    "media_id": getattr(req_msg, 'media_id', ''),
+                }
+            )
+
+            logger.info("WeCom App message forwarded to bus")
+
+        except Exception as e:
+            logger.error("Error in async message handling: {}", e)
+
+
+    async def _download_media(self, media_id: str, media_type: str) -> str | None:
+        """Download media from WeCom API and save to local file."""
+        if not media_id:
+            return None
+
+        token = await self._get_access_token()
+        if not token:
+            return None
+
+        # Create a fresh httpx client for this request to avoid event loop issues
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            try:
+                url = f"https://qyapi.weixin.qq.com/cgi-bin/media/get?access_token={token}&media_id={media_id}"
+                resp = await client.get(url)
+
+                # Check if response is JSON (error) or binary (success)
+                content_type = resp.headers.get("content-type", "")
+
+                if "application/json" in content_type:
+                    data = resp.json()
+                    if data.get("errcode") != 0:
+                        logger.error("WeCom App download media failed: {}", data.get("errmsg"))
+                        return None
+
+                # Determine filename from headers or generate one
+                content_disposition = resp.headers.get("content-disposition", "")
+                if "filename=" in content_disposition:
+                    # Extract filename from content-disposition header
+                    import re
+                    match = re.search(r'filename="?([^";]+)"?', content_disposition)
+                    if match:
+                        filename = match.group(1)
+                    else:
+                        filename = None
+                else:
+                    filename = None
+
+                if not filename:
+                    ext = ".jpg" if media_type == "image" else ".mp4" if media_type == "video" else ".amr"
+                    filename = f"{media_type}_{media_id[:16]}{ext}"
+
+                # Ensure media directory exists
+                if self._media_dir:
+                    self._media_dir.mkdir(parents=True, exist_ok=True)
+
+                    # Save file
+                    file_path = self._media_dir / filename
+                    with open(file_path, "wb") as f:
+                        f.write(resp.content)
+
+                    logger.info("WeCom App downloaded {} to {}", media_type, file_path)
+                    return str(file_path)
+
+            except Exception as e:
+                logger.error("Error downloading WeCom App media: {}", e)
+                return None
+
+    async def _get_access_token(self) -> str | None:
+        """Get or refresh Access Token for WeCom API."""
+        # Return cached token if valid
+        if self._access_token and time.time() < self._token_expiry:
+            return self._access_token
+
+        # Check if we have credentials
+        agent_id = getattr(self.config, 'agentid', None)
+        secret = getattr(self.config, 'secret', None)
+
+        if not agent_id:
+            logger.warning("WeCom App agent_id not configured")
+            return None
+        if not secret:
+            logger.warning("WeCom App secret not configured")
+            return None
+
+        # Use lock to prevent concurrent token refreshes
+        if self._token_lock:
+            async with self._token_lock:
+                # Double-check after acquiring lock
+                if self._access_token and time.time() < self._token_expiry:
+                    return self._access_token
+
+                # Use fresh httpx client to avoid event loop issues
+                try:
+                    async with httpx.AsyncClient(timeout=30.0) as client:
+                        url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.config.corp_id}&corpsecret={secret}"
+                        resp = await client.get(url)
+                        resp.raise_for_status()
+                        data = resp.json()
+
+                        if data.get("errcode") != 0:
+                            logger.error("WeCom App gettoken failed: {}", data.get("errmsg"))
+                            return None
+
+                        self._access_token = data.get("access_token")
+                        expires_in = data.get("expires_in", 7200)
+                        self._token_expiry = time.time() + expires_in - 60
+
+                        logger.info("WeCom App access token refreshed")
+                        return self._access_token
+
+                except Exception as e:
+                    logger.error("Error getting WeCom App access token: {}", e)
+                    return None
+        else:
+            # Fallback if lock not initialized - use fresh client
+            try:
+                async with httpx.AsyncClient(timeout=30.0) as client:
+                    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.config.corp_id}&corpsecret={secret}"
+                    resp = await client.get(url)
+                    resp.raise_for_status()
+                    data = resp.json()
+
+                    if data.get("errcode") != 0:
+                        logger.error("WeCom App gettoken failed: {}", data.get("errmsg"))
+                        return None
+
+                    self._access_token = data.get("access_token")
+                    expires_in = data.get("expires_in", 7200)
+                    self._token_expiry = time.time() + expires_in - 60
+
+                    logger.info("WeCom App access token refreshed")
+                    return self._access_token
+
+            except Exception as e:
+                logger.error("Error getting WeCom App access token: {}", e)
+                return None
+
+    async def _send_via_api(self, user_id: str, content: str) -> bool:
+        """Send message via WeCom API."""
+        token = await self._get_access_token()
+        if not token:
+            return False
+
+        # Create a fresh httpx client for this request to avoid event loop issues
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            try:
+                url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
+
+                payload = {
+                    "touser": user_id,
+                    "msgtype": "text",
+                    "agentid": getattr(self.config, 'agentid', ''),
+                    "text": {"content": content}
+                }
+
+                resp = await client.post(url, json=payload)
+                resp.raise_for_status()
+                data = resp.json()
+
+                if data.get("errcode") != 0:
+                    logger.error("WeCom App send failed: {}", data.get("errmsg"))
+                    return False
+
+                logger.info("WeCom App message sent via API to {}", user_id)
+                return True
+
+            except Exception as e:
+                logger.error("Error sending WeCom App message via API: {}", e)
+                return False
+
+    async def send(self, msg: OutboundMessage) -> None:
+        """Send a message through WeCom App."""
+        try:
+            content = msg.content.strip()
+            if not content:
+                return
+
+            # Check if we have API credentials
+            agent_id = getattr(self.config, 'agentid', None)
+            secret = getattr(self.config, 'secret', None)
+
+            if agent_id and secret:
+                user_id = msg.chat_id
+                success = await self._send_via_api(user_id, content)
+                if success:
+                    logger.info("WeCom App message sent to {}", msg.chat_id)
+                else:
+                    logger.warning("Failed to send WeCom App message to {}", msg.chat_id)
+            else:
+                logger.warning(
+                    "WeCom App agent_id/secret not configured. "
+                    "Cannot send proactive messages."
+                )
+
+        except Exception as e:
+            logger.error("Error sending WeCom App message: {}", e)

nanobot/channels/whatsapp.py

@@ -4,7 +4,7 @@ import asyncio
 import json
 import mimetypes
 from collections import OrderedDict
-from typing import Any
+from typing import Any, Literal
 
 from loguru import logger
 
@@ -23,6 +23,7 @@ class WhatsAppConfig(Base):
     bridge_url: str = "ws://localhost:3001"
     bridge_token: str = ""
     allow_from: list[str] = Field(default_factory=list)
+    group_policy: Literal["open", "mention"] = "open"  # "open" responds to all, "mention" only when @mentioned
 
 
 class WhatsAppChannel(BaseChannel):
@@ -138,6 +139,13 @@ class WhatsAppChannel(BaseChannel):
                     self._processed_message_ids.popitem(last=False)
 
             # Extract just the phone number or lid as chat_id
+            is_group = data.get("isGroup", False)
+            was_mentioned = data.get("wasMentioned", False)
+
+            if is_group and getattr(self.config, "group_policy", "open") == "mention":
+                if not was_mentioned:
+                    return
+
             user_id = pn if pn else sender
             sender_id = user_id.split("@")[0] if "@" in user_id else user_id
             logger.info("Sender {}", sender)

nanobot/cli/commands.py

@@ -1,11 +1,11 @@
 """CLI commands for nanobot."""
 
 import asyncio
-from contextlib import contextmanager, nullcontext
 import os
 import select
 import signal
 import sys
+from contextlib import contextmanager, nullcontext
 from pathlib import Path
 from typing import Any
 
@@ -21,12 +21,11 @@ if sys.platform == "win32":
             pass
 
 import typer
-from prompt_toolkit import print_formatted_text
+from prompt_toolkit import PromptSession, print_formatted_text
-from prompt_toolkit import PromptSession
+from prompt_toolkit.application import run_in_terminal
 from prompt_toolkit.formatted_text import ANSI, HTML
 from prompt_toolkit.history import FileHistory
 from prompt_toolkit.patch_stdout import patch_stdout
-from prompt_toolkit.application import run_in_terminal
 from rich.console import Console
 from rich.markdown import Markdown
 from rich.table import Table
@@ -65,6 +64,7 @@ def _flush_pending_tty_input() -> None:
 
     try:
         import termios
+
         termios.tcflush(fd, termios.TCIFLUSH)
         return
     except Exception:
@@ -87,6 +87,7 @@ def _restore_terminal() -> None:
         return
     try:
         import termios
+
         termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _SAVED_TERM_ATTRS)
     except Exception:
         pass
@@ -99,6 +100,7 @@ def _init_prompt_session() -> None:
     # Save terminal state so we can restore it on exit
     try:
         import termios
+
         _SAVED_TERM_ATTRS = termios.tcgetattr(sys.stdin.fileno())
     except Exception:
         pass
@@ -111,7 +113,7 @@ def _init_prompt_session() -> None:
     _PROMPT_SESSION = PromptSession(
         history=FileHistory(str(history_file)),
         enable_open_in_editor=False,
-        multiline=False,   # Enter submits (single line mode)
+        multiline=False,  # Enter submits (single line mode)
     )
 
 
@@ -144,10 +146,9 @@ def _print_agent_response(response: str, render_markdown: bool) -> None:
 
 async def _print_interactive_line(text: str) -> None:
     """Print async interactive updates with prompt_toolkit-safe Rich styling."""
+
     def _write() -> None:
-        ansi = _render_interactive_ansi(
+        ansi = _render_interactive_ansi(lambda c: c.print(f"  [dim]↳ {text}[/dim]"))
-            lambda c: c.print(f"  [dim]↳ {text}[/dim]")
-        )
         print_formatted_text(ANSI(ansi), end="")
 
     await run_in_terminal(_write)
@@ -155,6 +156,7 @@ async def _print_interactive_line(text: str) -> None:
 
 async def _print_interactive_response(response: str, render_markdown: bool) -> None:
     """Print async interactive replies with prompt_toolkit-safe Rich styling."""
+
     def _write() -> None:
         content = response or ""
         ansi = _render_interactive_ansi(
@@ -174,9 +176,9 @@ class _ThinkingSpinner:
     """Spinner wrapper with pause support for clean progress output."""
 
     def __init__(self, enabled: bool):
-        self._spinner = console.status(
+        self._spinner = (
-            "[dim]nanobot is thinking...[/dim]", spinner="dots"
+            console.status("[dim]nanobot is thinking...[/dim]", spinner="dots") if enabled else None
-        ) if enabled else None
+        )
         self._active = False
 
     def __enter__(self):
@@ -239,7 +241,6 @@ async def _read_interactive_input_async() -> str:
         raise KeyboardInterrupt from exc
 
 
-
 def version_callback(value: bool):
     if value:
         console.print(f"{__logo__} nanobot v{__version__}")
@@ -248,9 +249,7 @@ def version_callback(value: bool):
 
 @app.callback()
 def main(
-    version: bool = typer.Option(
+    version: bool = typer.Option(None, "--version", "-v", callback=version_callback, is_eager=True),
-        None, "--version", "-v", callback=version_callback, is_eager=True
-    ),
 ):
     """nanobot - Personal AI Assistant."""
     pass
@@ -265,6 +264,9 @@ def main(
 def onboard(
     workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
     config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+    non_interactive: bool = typer.Option(
+        False, "--non-interactive", help="Skip interactive wizard"
+    ),
 ):
     """Initialize nanobot configuration and workspace."""
     from nanobot.config.loader import get_config_path, load_config, save_config, set_config_path
@@ -282,45 +284,85 @@ def onboard(
             loaded.agents.defaults.workspace = workspace
         return loaded
 
-    # Create or update config
+    cfg: Config
-    if config_path.exists():
+
-        console.print(f"[yellow]Config already exists at {config_path}[/yellow]")
+    # Non-interactive mode: simple config creation/update
-        console.print("  [bold]y[/bold] = overwrite with defaults (existing values will be lost)")
+    if non_interactive:
-        console.print("  [bold]N[/bold] = refresh config, keeping existing values and adding new fields")
+        if config_path.exists():
-        if typer.confirm("Overwrite?"):
+            console.print(f"[yellow]Config already exists at {config_path}[/yellow]")
-            config = _apply_workspace_override(Config())
+            console.print(
-            save_config(config, config_path)
+                "  [bold]y[/bold] = overwrite with defaults (existing values will be lost)"
-            console.print(f"[green]✓[/green] Config reset to defaults at {config_path}")
+            )
+            console.print(
+                "  [bold]N[/bold] = refresh config, keeping existing values and adding new fields"
+            )
+            if typer.confirm("Overwrite?"):
+                cfg = _apply_workspace_override(Config())
+                save_config(cfg, config_path)
+                console.print(f"[green]✓[/green] Config reset to defaults at {config_path}")
+            else:
+                cfg = _apply_workspace_override(load_config(config_path))
+                save_config(cfg, config_path)
+                console.print(
+                    f"[green]✓[/green] Config refreshed at {config_path} (existing values preserved)"
+                )
         else:
-            config = _apply_workspace_override(load_config(config_path))
+            cfg = _apply_workspace_override(Config())
-            save_config(config, config_path)
+            save_config(cfg, config_path)
-            console.print(f"[green]✓[/green] Config refreshed at {config_path} (existing values preserved)")
+            console.print(f"[green]✓[/green] Created config at {config_path}")
+        console.print(
+            "[dim]Config template now uses `maxTokens` + `contextWindowTokens`; `memoryWindow` is no longer a runtime setting.[/dim]"
+        )
     else:
-        config = _apply_workspace_override(Config())
+        # Interactive mode: use wizard
-        save_config(config, config_path)
+        if config_path.exists():
-        console.print(f"[green]✓[/green] Created config at {config_path}")
+            cfg = _apply_workspace_override(load_config(config_path))
-    console.print("[dim]Config template now uses `maxTokens` + `contextWindowTokens`; `memoryWindow` is no longer a runtime setting.[/dim]")
+        else:
+            cfg = _apply_workspace_override(Config())
+
+        # Run interactive wizard
+        from nanobot.cli.onboard_wizard import run_onboard
+
+        try:
+            result = run_onboard(initial_config=cfg)
+            if not result.should_save:
+                console.print("[yellow]Configuration discarded. No changes were saved.[/yellow]")
+                return
+
+            cfg = result.config
+            save_config(cfg, config_path)
+            console.print(f"[green]✓[/green] Config saved at {config_path}")
+        except Exception as e:
+            console.print(f"[red]✗[/red] Error during configuration: {e}")
+            console.print("[yellow]Please run 'nanobot onboard' again to complete setup.[/yellow]")
+            raise typer.Exit(1)
 
     _onboard_plugins(config_path)
 
     # Create workspace, preferring the configured workspace path.
-    workspace = get_workspace_path(config.workspace_path)
+    workspace_path = get_workspace_path(cfg.workspace_path)
-    if not workspace.exists():
+    if not workspace_path.exists():
-        workspace.mkdir(parents=True, exist_ok=True)
+        workspace_path.mkdir(parents=True, exist_ok=True)
-        console.print(f"[green]✓[/green] Created workspace at {workspace}")
+        console.print(f"[green]✓[/green] Created workspace at {workspace_path}")
 
-    sync_workspace_templates(workspace)
+    sync_workspace_templates(workspace_path)
 
     agent_cmd = 'nanobot agent -m "Hello!"'
-    if config:
+    if cfg:
         agent_cmd += f" --config {config_path}"
 
     console.print(f"\n{__logo__} nanobot is ready!")
     console.print("\nNext steps:")
-    console.print(f"  1. Add your API key to [cyan]{config_path}[/cyan]")
+    if non_interactive:
-    console.print("     Get one at: https://openrouter.ai/keys")
+        console.print(f"  1. Add your API key to [cyan]{config_path}[/cyan]")
-    console.print(f"  2. Chat: [cyan]{agent_cmd}[/cyan]")
+        console.print("     Get one at: https://openrouter.ai/keys")
-    console.print("\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]")
+        console.print(f"  2. Chat: [cyan]{agent_cmd}[/cyan]")
+    else:
+        console.print('  1. Chat: [cyan]nanobot agent -m "Hello!"[/cyan]')
+        console.print("  2. Start gateway: [cyan]nanobot gateway[/cyan]")
+    console.print(
+        "\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]"
+    )
 
 
 def _merge_missing_defaults(existing: Any, defaults: Any) -> Any:
@@ -363,9 +405,9 @@ def _onboard_plugins(config_path: Path) -> None:
 
 def _make_provider(config: Config):
     """Create the appropriate LLM provider from config."""
+    from nanobot.providers.azure_openai_provider import AzureOpenAIProvider
     from nanobot.providers.base import GenerationSettings
     from nanobot.providers.openai_codex_provider import OpenAICodexProvider
-    from nanobot.providers.azure_openai_provider import AzureOpenAIProvider
 
     model = config.agents.defaults.model
     provider_name = config.get_provider_name(model)
@@ -377,6 +419,7 @@ def _make_provider(config: Config):
     # Custom: direct OpenAI-compatible endpoint, bypasses LiteLLM
     elif provider_name == "custom":
         from nanobot.providers.custom_provider import CustomProvider
+
         provider = CustomProvider(
             api_key=p.api_key if p else "no-key",
             api_base=config.get_api_base(model) or "http://localhost:8000/v1",
@@ -395,11 +438,25 @@ def _make_provider(config: Config):
             api_base=p.api_base,
             default_model=model,
         )
+    # OpenVINO Model Server: direct OpenAI-compatible endpoint at /v3
+    elif provider_name == "ovms":
+        from nanobot.providers.custom_provider import CustomProvider
+
+        provider = CustomProvider(
+            api_key=p.api_key if p else "no-key",
+            api_base=config.get_api_base(model) or "http://localhost:8000/v3",
+            default_model=model,
+        )
     else:
         from nanobot.providers.litellm_provider import LiteLLMProvider
         from nanobot.providers.registry import find_by_name
+
         spec = find_by_name(provider_name)
-        if not model.startswith("bedrock/") and not (p and p.api_key) and not (spec and (spec.is_oauth or spec.is_local)):
+        if (
+            not model.startswith("bedrock/")
+            and not (p and p.api_key)
+            and not (spec and (spec.is_oauth or spec.is_local))
+        ):
             console.print("[red]Error: No API key configured.[/red]")
             console.print("Set one in ~/.nanobot/config.json under providers section")
             raise typer.Exit(1)
@@ -473,6 +530,7 @@ def gateway(
 
     if verbose:
         import logging
+
         logging.basicConfig(level=logging.DEBUG)
 
     config = _load_runtime_config(config, workspace)
@@ -500,6 +558,7 @@ def gateway(
         web_search_config=config.tools.web.search,
         web_proxy=config.tools.web.proxy or None,
         exec_config=config.tools.exec,
+        input_limits=config.tools.input_limits,
         cron_service=cron,
         restrict_to_workspace=config.tools.restrict_to_workspace,
         session_manager=session_manager,
@@ -541,16 +600,23 @@ def gateway(
 
         if job.payload.deliver and job.payload.to and response:
             should_notify = await evaluate_response(
-                response, job.payload.message, provider, agent.model,
+                response,
+                job.payload.message,
+                provider,
+                agent.model,
             )
             if should_notify:
                 from nanobot.bus.events import OutboundMessage
-                await bus.publish_outbound(OutboundMessage(
+
-                    channel=job.payload.channel or "cli",
+                await bus.publish_outbound(
-                    chat_id=job.payload.to,
+                    OutboundMessage(
-                    content=response,
+                        channel=job.payload.channel or "cli",
-                ))
+                        chat_id=job.payload.to,
+                        content=response,
+                    )
+                )
         return response
+
     cron.on_job = on_cron_job
 
     # Create channel manager
@@ -591,10 +657,13 @@ def gateway(
     async def on_heartbeat_notify(response: str) -> None:
         """Deliver a heartbeat response to the user's channel."""
         from nanobot.bus.events import OutboundMessage
+
         channel, chat_id = _pick_heartbeat_target()
         if channel == "cli":
             return  # No external channel available to deliver to
-        await bus.publish_outbound(OutboundMessage(channel=channel, chat_id=chat_id, content=response))
+        await bus.publish_outbound(
+            OutboundMessage(channel=channel, chat_id=chat_id, content=response)
+        )
 
     hb_cfg = config.gateway.heartbeat
     heartbeat = HeartbeatService(
@@ -630,6 +699,7 @@ def gateway(
             console.print("\nShutting down...")
         except Exception:
             import traceback
+
             console.print("\n[red]Error: Gateway crashed unexpectedly[/red]")
             console.print(traceback.format_exc())
         finally:
@@ -642,8 +712,6 @@ def gateway(
     asyncio.run(run())
 
 
-
-
 # ============================================================================
 # Agent Commands
 # ============================================================================
@@ -655,8 +723,12 @@ def agent(
     session_id: str = typer.Option("cli:direct", "--session", "-s", help="Session ID"),
     workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
     config: str | None = typer.Option(None, "--config", "-c", help="Config file path"),
-    markdown: bool = typer.Option(True, "--markdown/--no-markdown", help="Render assistant output as Markdown"),
+    markdown: bool = typer.Option(
-    logs: bool = typer.Option(False, "--logs/--no-logs", help="Show nanobot runtime logs during chat"),
+        True, "--markdown/--no-markdown", help="Render assistant output as Markdown"
+    ),
+    logs: bool = typer.Option(
+        False, "--logs/--no-logs", help="Show nanobot runtime logs during chat"
+    ),
 ):
     """Interact with the agent directly."""
     from loguru import logger
@@ -692,6 +764,7 @@ def agent(
         web_search_config=config.tools.web.search,
         web_proxy=config.tools.web.proxy or None,
         exec_config=config.tools.exec,
+        input_limits=config.tools.input_limits,
         cron_service=cron,
         restrict_to_workspace=config.tools.restrict_to_workspace,
         mcp_servers=config.tools.mcp_servers,
@@ -715,7 +788,9 @@ def agent(
             nonlocal _thinking
             _thinking = _ThinkingSpinner(enabled=not logs)
             with _thinking:
-                response = await agent_loop.process_direct(message, session_id, on_progress=_cli_progress)
+                response = await agent_loop.process_direct(
+                    message, session_id, on_progress=_cli_progress
+                )
             _thinking = None
             _print_agent_response(response, render_markdown=markdown)
             await agent_loop.close_mcp()
@@ -724,8 +799,11 @@ def agent(
     else:
         # Interactive mode — route through bus like other channels
         from nanobot.bus.events import InboundMessage
+
         _init_prompt_session()
-        console.print(f"{__logo__} Interactive mode (type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit)\n")
+        console.print(
+            f"{__logo__} Interactive mode (type [bold]exit[/bold] or [bold]Ctrl+C[/bold] to quit)\n"
+        )
 
         if ":" in session_id:
             cli_channel, cli_chat_id = session_id.split(":", 1)
@@ -741,11 +819,11 @@ def agent(
         signal.signal(signal.SIGINT, _handle_signal)
         signal.signal(signal.SIGTERM, _handle_signal)
         # SIGHUP is not available on Windows
-        if hasattr(signal, 'SIGHUP'):
+        if hasattr(signal, "SIGHUP"):
             signal.signal(signal.SIGHUP, _handle_signal)
         # Ignore SIGPIPE to prevent silent process termination when writing to closed pipes
         # SIGPIPE is not available on Windows
-        if hasattr(signal, 'SIGPIPE'):
+        if hasattr(signal, "SIGPIPE"):
             signal.signal(signal.SIGPIPE, signal.SIG_IGN)
 
         async def run_interactive():
@@ -799,12 +877,14 @@ def agent(
                         turn_done.clear()
                         turn_response.clear()
 
-                        await bus.publish_inbound(InboundMessage(
+                        await bus.publish_inbound(
-                            channel=cli_channel,
+                            InboundMessage(
-                            sender_id="user",
+                                channel=cli_channel,
-                            chat_id=cli_chat_id,
+                                sender_id="user",
-                            content=user_input,
+                                chat_id=cli_chat_id,
-                        ))
+                                content=user_input,
+                            )
+                        )
 
                         nonlocal _thinking
                         _thinking = _ThinkingSpinner(enabled=not logs)
@@ -946,7 +1026,11 @@ def channels_login():
 
     env = {**os.environ}
     wa_cfg = getattr(config.channels, "whatsapp", None) or {}
-    bridge_token = wa_cfg.get("bridgeToken", "") if isinstance(wa_cfg, dict) else getattr(wa_cfg, "bridge_token", "")
+    bridge_token = (
+        wa_cfg.get("bridgeToken", "")
+        if isinstance(wa_cfg, dict)
+        else getattr(wa_cfg, "bridge_token", "")
+    )
     if bridge_token:
         env["BRIDGE_TOKEN"] = bridge_token
     env["AUTH_DIR"] = str(get_runtime_subdir("whatsapp-auth"))
@@ -1020,8 +1104,12 @@ def status():
 
     console.print(f"{__logo__} nanobot Status\n")
 
-    console.print(f"Config: {config_path} {'[green]✓[/green]' if config_path.exists() else '[red]✗[/red]'}")
+    console.print(
-    console.print(f"Workspace: {workspace} {'[green]✓[/green]' if workspace.exists() else '[red]✗[/red]'}")
+        f"Config: {config_path} {'[green]✓[/green]' if config_path.exists() else '[red]✗[/red]'}"
+    )
+    console.print(
+        f"Workspace: {workspace} {'[green]✓[/green]' if workspace.exists() else '[red]✗[/red]'}"
+    )
 
     if config_path.exists():
         from nanobot.providers.registry import PROVIDERS
@@ -1043,7 +1131,9 @@ def status():
                     console.print(f"{spec.label}: [dim]not set[/dim]")
             else:
                 has_key = bool(p.api_key)
-                console.print(f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}")
+                console.print(
+                    f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}"
+                )
 
 
 # ============================================================================
@@ -1061,12 +1151,15 @@ def _register_login(name: str):
     def decorator(fn):
         _LOGIN_HANDLERS[name] = fn
         return fn
+
     return decorator
 
 
 @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')"
+    ),
 ):
     """Authenticate with an OAuth provider."""
     from nanobot.providers.registry import PROVIDERS
@@ -1091,6 +1184,7 @@ def provider_login(
 def _login_openai_codex() -> None:
     try:
         from oauth_cli_kit import get_token, login_oauth_interactive
+
         token = None
         try:
             token = get_token()
@@ -1105,7 +1199,9 @@ def _login_openai_codex() -> None:
         if not (token and token.access):
             console.print("[red]✗ Authentication failed[/red]")
             raise typer.Exit(1)
-        console.print(f"[green]✓ Authenticated with OpenAI Codex[/green]  [dim]{token.account_id}[/dim]")
+        console.print(
+            f"[green]✓ Authenticated with OpenAI Codex[/green]  [dim]{token.account_id}[/dim]"
+        )
     except ImportError:
         console.print("[red]oauth_cli_kit not installed. Run: pip install oauth-cli-kit[/red]")
         raise typer.Exit(1)
@@ -1119,7 +1215,12 @@ def _login_github_copilot() -> None:
 
     async def _trigger():
         from litellm import acompletion
-        await acompletion(model="github_copilot/gpt-4o", messages=[{"role": "user", "content": "hi"}], max_tokens=1)
+
+        await acompletion(
+            model="github_copilot/gpt-4o",
+            messages=[{"role": "user", "content": "hi"}],
+            max_tokens=1,
+        )
 
     try:
         asyncio.run(_trigger())

nanobot/cli/model_info.py

@@ -0,0 +1,226 @@
+"""Model information helpers for the onboard wizard.
+
+Provides model context window lookup and autocomplete suggestions using litellm.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import Any
+
+import litellm
+
+
+@lru_cache(maxsize=1)
+def _get_model_cost_map() -> dict[str, Any]:
+    """Get litellm's model cost map (cached)."""
+    return getattr(litellm, "model_cost", {})
+
+
+@lru_cache(maxsize=1)
+def get_all_models() -> list[str]:
+    """Get all known model names from litellm.
+    """
+    models = set()
+
+    # From model_cost (has pricing info)
+    cost_map = _get_model_cost_map()
+    for k in cost_map.keys():
+        if k != "sample_spec":
+            models.add(k)
+
+    # From models_by_provider (more complete provider coverage)
+    for provider_models in getattr(litellm, "models_by_provider", {}).values():
+        if isinstance(provider_models, (set, list)):
+            models.update(provider_models)
+
+    return sorted(models)
+
+
+def _normalize_model_name(model: str) -> str:
+    """Normalize model name for comparison."""
+    return model.lower().replace("-", "_").replace(".", "")
+
+
+def find_model_info(model_name: str) -> dict[str, Any] | None:
+    """Find model info with fuzzy matching.
+
+    Args:
+        model_name: Model name in any common format
+
+    Returns:
+        Model info dict or None if not found
+    """
+    cost_map = _get_model_cost_map()
+    if not cost_map:
+        return None
+
+    # Direct match
+    if model_name in cost_map:
+        return cost_map[model_name]
+
+    # Extract base name (without provider prefix)
+    base_name = model_name.split("/")[-1] if "/" in model_name else model_name
+    base_normalized = _normalize_model_name(base_name)
+
+    candidates = []
+
+    for key, info in cost_map.items():
+        if key == "sample_spec":
+            continue
+
+        key_base = key.split("/")[-1] if "/" in key else key
+        key_base_normalized = _normalize_model_name(key_base)
+
+        # Score the match
+        score = 0
+
+        # Exact base name match (highest priority)
+        if base_normalized == key_base_normalized:
+            score = 100
+        # Base name contains model
+        elif base_normalized in key_base_normalized:
+            score = 80
+        # Model contains base name
+        elif key_base_normalized in base_normalized:
+            score = 70
+        # Partial match
+        elif base_normalized[:10] in key_base_normalized:
+            score = 50
+
+        if score > 0:
+            # Prefer models with max_input_tokens
+            if info.get("max_input_tokens"):
+                score += 10
+            candidates.append((score, key, info))
+
+    if not candidates:
+        return None
+
+    # Return the best match
+    candidates.sort(key=lambda x: (-x[0], x[1]))
+    return candidates[0][2]
+
+
+def get_model_context_limit(model: str, provider: str = "auto") -> int | None:
+    """Get the maximum input context tokens for a model.
+
+    Args:
+        model: Model name (e.g., "claude-3.5-sonnet", "gpt-4o")
+        provider: Provider name for informational purposes (not yet used for filtering)
+
+    Returns:
+        Maximum input tokens, or None if unknown
+
+    Note:
+        The provider parameter is currently informational only. Future versions may
+        use it to prefer provider-specific model variants in the lookup.
+    """
+    # First try fuzzy search in model_cost (has more accurate max_input_tokens)
+    info = find_model_info(model)
+    if info:
+        # Prefer max_input_tokens (this is what we want for context window)
+        max_input = info.get("max_input_tokens")
+        if max_input and isinstance(max_input, int):
+            return max_input
+
+    # Fall back to litellm's get_max_tokens (returns max_output_tokens typically)
+    try:
+        result = litellm.get_max_tokens(model)
+        if result and result > 0:
+            return result
+    except (KeyError, ValueError, AttributeError):
+        # Model not found in litellm's database or invalid response
+        pass
+
+    # Last resort: use max_tokens from model_cost
+    if info:
+        max_tokens = info.get("max_tokens")
+        if max_tokens and isinstance(max_tokens, int):
+            return max_tokens
+
+    return None
+
+
+@lru_cache(maxsize=1)
+def _get_provider_keywords() -> dict[str, list[str]]:
+    """Build provider keywords mapping from nanobot's provider registry.
+
+    Returns:
+        Dict mapping provider name to list of keywords for model filtering.
+    """
+    try:
+        from nanobot.providers.registry import PROVIDERS
+
+        mapping = {}
+        for spec in PROVIDERS:
+            if spec.keywords:
+                mapping[spec.name] = list(spec.keywords)
+        return mapping
+    except ImportError:
+        return {}
+
+
+def get_model_suggestions(partial: str, provider: str = "auto", limit: int = 20) -> list[str]:
+    """Get autocomplete suggestions for model names.
+
+    Args:
+        partial: Partial model name typed by user
+        provider: Provider name for filtering (e.g., "openrouter", "minimax")
+        limit: Maximum number of suggestions to return
+
+    Returns:
+        List of matching model names
+    """
+    all_models = get_all_models()
+    if not all_models:
+        return []
+
+    partial_lower = partial.lower()
+    partial_normalized = _normalize_model_name(partial)
+
+    # Get provider keywords from registry
+    provider_keywords = _get_provider_keywords()
+
+    # Filter by provider if specified
+    allowed_keywords = None
+    if provider and provider != "auto":
+        allowed_keywords = provider_keywords.get(provider.lower())
+
+    matches = []
+
+    for model in all_models:
+        model_lower = model.lower()
+
+        # Apply provider filter
+        if allowed_keywords:
+            if not any(kw in model_lower for kw in allowed_keywords):
+                continue
+
+        # Match against partial input
+        if not partial:
+            matches.append(model)
+            continue
+
+        if partial_lower in model_lower:
+            # Score by position of match (earlier = better)
+            pos = model_lower.find(partial_lower)
+            score = 100 - pos
+            matches.append((score, model))
+        elif partial_normalized in _normalize_model_name(model):
+            score = 50
+            matches.append((score, model))
+
+    # Sort by score if we have scored matches
+    if matches and isinstance(matches[0], tuple):
+        matches.sort(key=lambda x: (-x[0], x[1]))
+        matches = [m[1] for m in matches]
+    else:
+        matches.sort()
+
+    return matches[:limit]
+
+
+def format_token_count(tokens: int) -> str:
+    """Format token count for display (e.g., 200000 -> '200,000')."""
+    return f"{tokens:,}"

nanobot/config/loader.py

@@ -3,6 +3,9 @@
 import json
 from pathlib import Path
 
+import pydantic
+from loguru import logger
+
 from nanobot.config.schema import Config
 
 
@@ -41,9 +44,9 @@ def load_config(config_path: Path | None = None) -> Config:
                 data = json.load(f)
             data = _migrate_config(data)
             return Config.model_validate(data)
-        except (json.JSONDecodeError, ValueError) as e:
+        except (json.JSONDecodeError, ValueError, pydantic.ValidationError) as e:
-            print(f"Warning: Failed to load config from {path}: {e}")
+            logger.warning(f"Failed to load config from {path}: {e}")
-            print("Using default configuration.")
+            logger.warning("Using default configuration.")
 
     return Config()
 

nanobot/config/schema.py

@@ -76,9 +76,11 @@ class ProvidersConfig(Base):
     dashscope: ProviderConfig = Field(default_factory=ProviderConfig)
     vllm: ProviderConfig = Field(default_factory=ProviderConfig)
     ollama: ProviderConfig = Field(default_factory=ProviderConfig)  # Ollama local models
+    ovms: ProviderConfig = Field(default_factory=ProviderConfig)  # OpenVINO Model Server (OVMS)
     gemini: ProviderConfig = Field(default_factory=ProviderConfig)
     moonshot: ProviderConfig = Field(default_factory=ProviderConfig)
     minimax: ProviderConfig = Field(default_factory=ProviderConfig)
+    mistral: ProviderConfig = Field(default_factory=ProviderConfig)
     aihubmix: ProviderConfig = Field(default_factory=ProviderConfig)  # AiHubMix API gateway
     siliconflow: ProviderConfig = Field(default_factory=ProviderConfig)  # SiliconFlow (硅基流动)
     volcengine: ProviderConfig = Field(default_factory=ProviderConfig)  # VolcEngine (火山引擎)
@@ -129,6 +131,13 @@ class ExecToolConfig(Base):
     path_append: str = ""
 
 
+class InputLimitsConfig(Base):
+    """Limits for user-provided multimodal inputs."""
+
+    max_input_images: int = 3
+    max_input_image_bytes: int = 10 * 1024 * 1024
+
+
 class MCPServerConfig(Base):
     """MCP server connection configuration (stdio or HTTP)."""
 
@@ -146,6 +155,7 @@ class ToolsConfig(Base):
 
     web: WebToolsConfig = Field(default_factory=WebToolsConfig)
     exec: ExecToolConfig = Field(default_factory=ExecToolConfig)
+    input_limits: InputLimitsConfig = Field(default_factory=InputLimitsConfig)
     restrict_to_workspace: bool = False  # If true, restrict all tool access to workspace directory
     mcp_servers: dict[str, MCPServerConfig] = Field(default_factory=dict)
 

nanobot/providers/registry.py

@@ -399,6 +399,23 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         strip_model_prefix=False,
         model_overrides=(),
     ),
+    # Mistral AI: OpenAI-compatible API at api.mistral.ai/v1.
+    ProviderSpec(
+        name="mistral",
+        keywords=("mistral",),
+        env_key="MISTRAL_API_KEY",
+        display_name="Mistral",
+        litellm_prefix="mistral",  # mistral-large-latest → mistral/mistral-large-latest
+        skip_prefixes=("mistral/",),  # avoid double-prefix
+        env_extras=(),
+        is_gateway=False,
+        is_local=False,
+        detect_by_key_prefix="",
+        detect_by_base_keyword="",
+        default_api_base="https://api.mistral.ai/v1",
+        strip_model_prefix=False,
+        model_overrides=(),
+    ),
     # === Local deployment (matched by config key, NOT by api_base) =========
     # vLLM / any OpenAI-compatible local server.
     # Detected when config key is "vllm" (provider_name="vllm").
@@ -435,6 +452,17 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         strip_model_prefix=False,
         model_overrides=(),
     ),
+    # === OpenVINO Model Server (direct, local, OpenAI-compatible at /v3) ===
+    ProviderSpec(
+        name="ovms",
+        keywords=("openvino", "ovms"),
+        env_key="",
+        display_name="OpenVINO Model Server",
+        litellm_prefix="",
+        is_direct=True,
+        is_local=True,
+        default_api_base="http://localhost:8000/v3",
+    ),
     # === Auxiliary (not a primary LLM provider) ============================
     # Groq: mainly used for Whisper voice transcription, also usable for LLM.
     # Needs "groq/" prefix for LiteLLM routing. Placed last — it rarely wins fallback.

nanobot/skills/cron/SKILL.md

@@ -30,6 +30,11 @@ One-time scheduled task (compute ISO datetime from current time):
 cron(action="add", message="Remind me about the meeting", at="<ISO datetime>")
 ```
 
+One-time task with timezone (naive datetime interpreted in given tz):
+```
+cron(action="add", message="Drink water!", at="2026-03-18T14:40:00", tz="Asia/Shanghai")
+```
+
 Timezone-aware cron:
 ```
 cron(action="add", message="Morning standup", cron_expr="0 9 * * 1-5", tz="America/Vancouver")
@@ -51,7 +56,8 @@ cron(action="remove", job_id="abc123")
 | weekdays at 5pm | cron_expr: "0 17 * * 1-5" |
 | 9am Vancouver time daily | cron_expr: "0 9 * * *", tz: "America/Vancouver" |
 | at a specific time | at: ISO datetime string (compute from current time) |
+| at 2pm Shanghai time | at: "2026-03-18T14:00:00", tz: "Asia/Shanghai" |
 
 ## Timezone
 
-Use `tz` with `cron_expr` to schedule in a specific IANA timezone. Without `tz`, the server's local timezone is used.
+Use `tz` with `cron_expr` or `at` to schedule in a specific IANA timezone. Without `tz`, the server's local timezone is used.

pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "qq-botpy>=1.2.0,<2.0.0",
     "python-socks[asyncio]>=2.8.0,<3.0.0",
     "prompt-toolkit>=3.0.50,<4.0.0",
+    "questionary>=2.0.0,<3.0.0",
     "mcp>=1.26.0,<2.0.0",
     "json-repair>=0.57.0,<1.0.0",
     "chardet>=3.0.2,<6.0.0",
@@ -53,6 +54,9 @@ dependencies = [
 wecom = [
     "wecom-aibot-sdk-python>=0.1.5",
 ]
+wecom-app-svr = [
+    "wecom-app-svr>=0.1.0",
+]
 matrix = [
     "matrix-nio[e2e]>=0.25.2",
     "mistune>=3.0.0,<4.0.0",

tests/test_commands.py

@@ -13,27 +13,28 @@ from nanobot.providers.litellm_provider import LiteLLMProvider
 from nanobot.providers.openai_codex_provider import _strip_model_prefix
 from nanobot.providers.registry import find_by_model
 
-
-def _strip_ansi(text):
-    """Remove ANSI escape codes from text."""
-    ansi_escape = re.compile(r'\x1b\[[0-9;]*m')
-    return ansi_escape.sub('', text)
-
 runner = CliRunner()
 
 
-class _StopGateway(RuntimeError):
+class _StopGatewayError(RuntimeError):
     pass
 
 
+def _strip_ansi(text):
+    """Remove ANSI escape codes from text."""
+    ansi_escape = re.compile(r"\x1b\[[0-9;]*m")
+    return ansi_escape.sub("", text)
+
+
 @pytest.fixture
 def mock_paths():
     """Mock config/workspace paths for test isolation."""
-    with patch("nanobot.config.loader.get_config_path") as mock_cp, \
+    with (
-         patch("nanobot.config.loader.save_config") as mock_sc, \
+        patch("nanobot.config.loader.get_config_path") as mock_cp,
-         patch("nanobot.config.loader.load_config") as mock_lc, \
+        patch("nanobot.config.loader.save_config") as mock_sc,
-         patch("nanobot.cli.commands.get_workspace_path") as mock_ws:
+        patch("nanobot.config.loader.load_config") as mock_lc,
-
+        patch("nanobot.cli.commands.get_workspace_path") as mock_ws,
+    ):
         base_dir = Path("./test_onboard_data")
         if base_dir.exists():
             shutil.rmtree(base_dir)
@@ -59,11 +60,11 @@ def mock_paths():
             shutil.rmtree(base_dir)
 
 
-def test_onboard_fresh_install(mock_paths):
+def test_onboard_fresh_install_non_interactive(mock_paths):
-    """No existing config — should create from scratch."""
+    """No existing config — should create from scratch in non-interactive mode."""
     config_file, workspace_dir, mock_ws = mock_paths
 
-    result = runner.invoke(app, ["onboard"])
+    result = runner.invoke(app, ["onboard", "--non-interactive"])
 
     assert result.exit_code == 0
     assert "Created config" in result.stdout
@@ -76,12 +77,12 @@ def test_onboard_fresh_install(mock_paths):
     assert mock_ws.call_args.args == (expected_workspace,)
 
 
-def test_onboard_existing_config_refresh(mock_paths):
+def test_onboard_existing_config_refresh_non_interactive(mock_paths):
     """Config exists, user declines overwrite — should refresh (load-merge-save)."""
     config_file, workspace_dir, _ = mock_paths
     config_file.write_text('{"existing": true}')
 
-    result = runner.invoke(app, ["onboard"], input="n\n")
+    result = runner.invoke(app, ["onboard", "--non-interactive"], input="n\n")
 
     assert result.exit_code == 0
     assert "Config already exists" in result.stdout
@@ -90,12 +91,12 @@ def test_onboard_existing_config_refresh(mock_paths):
     assert (workspace_dir / "AGENTS.md").exists()
 
 
-def test_onboard_existing_config_overwrite(mock_paths):
+def test_onboard_existing_config_overwrite_non_interactive(mock_paths):
     """Config exists, user confirms overwrite — should reset to defaults."""
     config_file, workspace_dir, _ = mock_paths
     config_file.write_text('{"existing": true}')
 
-    result = runner.invoke(app, ["onboard"], input="y\n")
+    result = runner.invoke(app, ["onboard", "--non-interactive"], input="y\n")
 
     assert result.exit_code == 0
     assert "Config already exists" in result.stdout
@@ -103,13 +104,13 @@ def test_onboard_existing_config_overwrite(mock_paths):
     assert workspace_dir.exists()
 
 
-def test_onboard_existing_workspace_safe_create(mock_paths):
+def test_onboard_existing_workspace_safe_create_non_interactive(mock_paths):
     """Workspace exists — should not recreate, but still add missing templates."""
     config_file, workspace_dir, _ = mock_paths
     workspace_dir.mkdir(parents=True)
     config_file.write_text("{}")
 
-    result = runner.invoke(app, ["onboard"], input="n\n")
+    result = runner.invoke(app, ["onboard", "--non-interactive"], input="n\n")
 
     assert result.exit_code == 0
     assert "Created workspace" not in result.stdout
@@ -126,9 +127,28 @@ def test_onboard_help_shows_workspace_and_config_options():
     assert "-w" in stripped_output
     assert "--config" in stripped_output
     assert "-c" in stripped_output
+    assert "--non-interactive" in stripped_output
     assert "--dir" not in stripped_output
 
 
+def test_onboard_interactive_discard_does_not_save_or_create_workspace(mock_paths, monkeypatch):
+    config_file, workspace_dir, _ = mock_paths
+
+    from nanobot.cli.onboard_wizard import OnboardResult
+
+    monkeypatch.setattr(
+        "nanobot.cli.onboard_wizard.run_onboard",
+        lambda initial_config: OnboardResult(config=initial_config, should_save=False),
+    )
+
+    result = runner.invoke(app, ["onboard"])
+
+    assert result.exit_code == 0
+    assert "No changes were saved" in result.stdout
+    assert not config_file.exists()
+    assert not workspace_dir.exists()
+
+
 def test_onboard_uses_explicit_config_and_workspace_paths(tmp_path, monkeypatch):
     config_path = tmp_path / "instance" / "config.json"
     workspace_path = tmp_path / "workspace"
@@ -137,7 +157,14 @@ def test_onboard_uses_explicit_config_and_workspace_paths(tmp_path, monkeypatch)
 
     result = runner.invoke(
         app,
-        ["onboard", "--config", str(config_path), "--workspace", str(workspace_path)],
+        [
+            "onboard",
+            "--config",
+            str(config_path),
+            "--workspace",
+            str(workspace_path),
+            "--non-interactive",
+        ],
     )
 
     assert result.exit_code == 0
@@ -277,15 +304,16 @@ def mock_agent_runtime(tmp_path):
     config.agents.defaults.workspace = str(tmp_path / "default-workspace")
     cron_dir = tmp_path / "data" / "cron"
 
-    with patch("nanobot.config.loader.load_config", return_value=config) as mock_load_config, \
+    with (
-         patch("nanobot.config.paths.get_cron_dir", return_value=cron_dir), \
+        patch("nanobot.config.loader.load_config", return_value=config) as mock_load_config,
-         patch("nanobot.cli.commands.sync_workspace_templates") as mock_sync_templates, \
+        patch("nanobot.config.paths.get_cron_dir", return_value=cron_dir),
-         patch("nanobot.cli.commands._make_provider", return_value=object()), \
+        patch("nanobot.cli.commands.sync_workspace_templates") as mock_sync_templates,
-         patch("nanobot.cli.commands._print_agent_response") as mock_print_response, \
+        patch("nanobot.cli.commands._make_provider", return_value=object()),
-         patch("nanobot.bus.queue.MessageBus"), \
+        patch("nanobot.cli.commands._print_agent_response") as mock_print_response,
-         patch("nanobot.cron.service.CronService"), \
+        patch("nanobot.bus.queue.MessageBus"),
-         patch("nanobot.agent.loop.AgentLoop") as mock_agent_loop_cls:
+        patch("nanobot.cron.service.CronService"),
-
+        patch("nanobot.agent.loop.AgentLoop") as mock_agent_loop_cls,
+    ):
         agent_loop = MagicMock()
         agent_loop.channels_config = None
         agent_loop.process_direct = AsyncMock(return_value="mock-response")
@@ -325,7 +353,9 @@ def test_agent_uses_default_config_when_no_workspace_or_config_flags(mock_agent_
         mock_agent_runtime["config"].workspace_path
     )
     mock_agent_runtime["agent_loop"].process_direct.assert_awaited_once()
-    mock_agent_runtime["print_response"].assert_called_once_with("mock-response", render_markdown=True)
+    mock_agent_runtime["print_response"].assert_called_once_with(
+        "mock-response", render_markdown=True
+    )
 
 
 def test_agent_uses_explicit_config_path(mock_agent_runtime, tmp_path: Path):
@@ -368,7 +398,9 @@ def test_agent_config_sets_active_path(monkeypatch, tmp_path: Path) -> None:
             return None
 
     monkeypatch.setattr("nanobot.agent.loop.AgentLoop", _FakeAgentLoop)
-    monkeypatch.setattr("nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        "nanobot.cli.commands._print_agent_response", lambda *_args, **_kwargs: None
+    )
 
     result = runner.invoke(app, ["agent", "-m", "hello", "-c", str(config_file)])
 
@@ -434,12 +466,12 @@ def test_gateway_uses_workspace_from_config_by_default(monkeypatch, tmp_path: Pa
     )
     monkeypatch.setattr(
         "nanobot.cli.commands._make_provider",
-        lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
+        lambda _config: (_ for _ in ()).throw(_StopGatewayError("stop")),
     )
 
     result = runner.invoke(app, ["gateway", "--config", str(config_file)])
 
-    assert isinstance(result.exception, _StopGateway)
+    assert isinstance(result.exception, _StopGatewayError)
     assert seen["config_path"] == config_file.resolve()
     assert seen["workspace"] == Path(config.agents.defaults.workspace)
 
@@ -462,7 +494,7 @@ def test_gateway_workspace_option_overrides_config(monkeypatch, tmp_path: Path)
     )
     monkeypatch.setattr(
         "nanobot.cli.commands._make_provider",
-        lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
+        lambda _config: (_ for _ in ()).throw(_StopGatewayError("stop")),
     )
 
     result = runner.invoke(
@@ -470,7 +502,7 @@ def test_gateway_workspace_option_overrides_config(monkeypatch, tmp_path: Path)
         ["gateway", "--config", str(config_file), "--workspace", str(override)],
     )
 
-    assert isinstance(result.exception, _StopGateway)
+    assert isinstance(result.exception, _StopGatewayError)
     assert seen["workspace"] == override
     assert config.workspace_path == override
 
@@ -488,15 +520,16 @@ def test_gateway_warns_about_deprecated_memory_window(monkeypatch, tmp_path: Pat
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr(
         "nanobot.cli.commands._make_provider",
-        lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
+        lambda _config: (_ for _ in ()).throw(_StopGatewayError("stop")),
     )
 
     result = runner.invoke(app, ["gateway", "--config", str(config_file)])
 
-    assert isinstance(result.exception, _StopGateway)
+    assert isinstance(result.exception, _StopGatewayError)
     assert "memoryWindow" in result.stdout
     assert "contextWindowTokens" in result.stdout
 
+
 def test_gateway_uses_config_directory_for_cron_store(monkeypatch, tmp_path: Path) -> None:
     config_file = tmp_path / "instance" / "config.json"
     config_file.parent.mkdir(parents=True)
@@ -517,13 +550,13 @@ def test_gateway_uses_config_directory_for_cron_store(monkeypatch, tmp_path: Pat
     class _StopCron:
         def __init__(self, store_path: Path) -> None:
             seen["cron_store"] = store_path
-            raise _StopGateway("stop")
+            raise _StopGatewayError("stop")
 
     monkeypatch.setattr("nanobot.cron.service.CronService", _StopCron)
 
     result = runner.invoke(app, ["gateway", "--config", str(config_file)])
 
-    assert isinstance(result.exception, _StopGateway)
+    assert isinstance(result.exception, _StopGatewayError)
     assert seen["cron_store"] == config_file.parent / "cron" / "jobs.json"
 
 
@@ -540,12 +573,12 @@ def test_gateway_uses_configured_port_when_cli_flag_is_missing(monkeypatch, tmp_
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr(
         "nanobot.cli.commands._make_provider",
-        lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
+        lambda _config: (_ for _ in ()).throw(_StopGatewayError("stop")),
     )
 
     result = runner.invoke(app, ["gateway", "--config", str(config_file)])
 
-    assert isinstance(result.exception, _StopGateway)
+    assert isinstance(result.exception, _StopGatewayError)
     assert "port 18791" in result.stdout
 
 
@@ -562,10 +595,10 @@ def test_gateway_cli_port_overrides_configured_port(monkeypatch, tmp_path: Path)
     monkeypatch.setattr("nanobot.cli.commands.sync_workspace_templates", lambda _path: None)
     monkeypatch.setattr(
         "nanobot.cli.commands._make_provider",
-        lambda _config: (_ for _ in ()).throw(_StopGateway("stop")),
+        lambda _config: (_ for _ in ()).throw(_StopGatewayError("stop")),
     )
 
     result = runner.invoke(app, ["gateway", "--config", str(config_file), "--port", "18792"])
 
-    assert isinstance(result.exception, _StopGateway)
+    assert isinstance(result.exception, _StopGatewayError)
     assert "port 18792" in result.stdout

tests/test_config_migration.py

@@ -78,7 +78,7 @@ def test_onboard_refresh_rewrites_legacy_config_template(tmp_path, monkeypatch)
     monkeypatch.setattr("nanobot.config.loader.get_config_path", lambda: config_path)
     monkeypatch.setattr("nanobot.cli.commands.get_workspace_path", lambda _workspace=None: workspace)
 
-    result = runner.invoke(app, ["onboard"], input="n\n")
+    result = runner.invoke(app, ["onboard", "--non-interactive"], input="n\n")
 
     assert result.exit_code == 0
     assert "contextWindowTokens" in result.stdout
@@ -125,7 +125,7 @@ def test_onboard_refresh_backfills_missing_channel_fields(tmp_path, monkeypatch)
         },
     )
 
-    result = runner.invoke(app, ["onboard"], input="n\n")
+    result = runner.invoke(app, ["onboard", "--non-interactive"], input="n\n")
 
     assert result.exit_code == 0
     saved = json.loads(config_path.read_text(encoding="utf-8"))

tests/test_context_multimodal.py

@@ -0,0 +1,106 @@
+from pathlib import Path
+
+from nanobot.agent.context import ContextBuilder
+from nanobot.config.schema import InputLimitsConfig
+
+
+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"
+)
+
+
+def _builder(tmp_path: Path, input_limits: InputLimitsConfig | None = None) -> ContextBuilder:
+    return ContextBuilder(tmp_path, input_limits=input_limits)
+
+
+def test_build_user_content_keeps_only_first_three_images(tmp_path: Path) -> None:
+    builder = _builder(tmp_path)
+    max_images = builder.input_limits.max_input_images
+    paths = []
+    for i in range(max_images + 1):
+        path = tmp_path / f"img{i}.png"
+        path.write_bytes(PNG_BYTES)
+        paths.append(str(path))
+
+    content = builder._build_user_content("describe these", paths)
+
+    assert isinstance(content, list)
+    assert sum(1 for block in content if block.get("type") == "image_url") == max_images
+    assert content[-1]["text"].startswith(
+        f"[Skipped 1 image: only the first {max_images} images are included]"
+    )
+
+
+def test_build_user_content_skips_invalid_images_with_note(tmp_path: Path) -> None:
+    builder = _builder(tmp_path)
+    bad = tmp_path / "not-image.txt"
+    bad.write_text("hello", encoding="utf-8")
+
+    content = builder._build_user_content("what is this?", [str(bad)])
+
+    assert isinstance(content, str)
+    assert "[Skipped image: unsupported or invalid image format (not-image.txt)]" in content
+    assert content.endswith("what is this?")
+
+
+def test_build_user_content_skips_missing_file(tmp_path: Path) -> None:
+    builder = _builder(tmp_path)
+
+    content = builder._build_user_content("hello", [str(tmp_path / "ghost.png")])
+
+    assert isinstance(content, str)
+    assert "[Skipped image: file not found (ghost.png)]" in content
+    assert content.endswith("hello")
+
+
+def test_build_user_content_skips_large_images_with_note(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)])
+
+    limit_mb = builder.input_limits.max_input_image_bytes // (1024 * 1024)
+    assert isinstance(content, str)
+    assert f"[Skipped image: file too large (big.png, limit {limit_mb} MB)]" in content
+
+
+def test_build_user_content_respects_custom_input_limits(tmp_path: Path) -> None:
+    builder = _builder(
+        tmp_path,
+        input_limits=InputLimitsConfig(max_input_images=1, max_input_image_bytes=1024),
+    )
+    small = tmp_path / "small.png"
+    large = tmp_path / "large.png"
+    small.write_bytes(PNG_BYTES)
+    large.write_bytes(PNG_BYTES + b"x" * 1024)
+
+    content = builder._build_user_content("describe", [str(small), str(large)])
+
+    assert isinstance(content, list)
+    assert sum(1 for block in content if block.get("type") == "image_url") == 1
+    assert content[-1]["text"].startswith("[Skipped 1 image: only the first 1 images are included]")
+
+
+def test_build_user_content_keeps_valid_images_and_skip_notes_together(tmp_path: Path) -> None:
+    builder = _builder(tmp_path)
+    good = tmp_path / "good.png"
+    bad = tmp_path / "bad.txt"
+    good.write_bytes(PNG_BYTES)
+    bad.write_text("oops", encoding="utf-8")
+
+    content = builder._build_user_content("check both", [str(good), str(bad)])
+
+    assert isinstance(content, list)
+    assert content[0]["type"] == "image_url"
+    assert (
+        "[Skipped image: unsupported or invalid image format (bad.txt)]"
+        in content[-1]["text"]
+    )
+    assert content[-1]["text"].endswith("check both")

tests/test_cron_tool_tz_at.py

@@ -0,0 +1,103 @@
+"""Tests for CronTool at+tz timezone handling."""
+
+from datetime import datetime, timezone
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from nanobot.agent.tools.cron import CronTool
+from nanobot.cron.service import CronService
+
+
+def _make_tool(tmp_path) -> CronTool:
+    service = CronService(tmp_path / "cron" / "jobs.json")
+    tool = CronTool(service)
+    tool.set_context("test-channel", "test-chat")
+    return tool
+
+
+@pytest.mark.asyncio
+async def test_at_with_tz_naive_datetime(tmp_path) -> None:
+    """Naive datetime + tz should be interpreted in the given timezone."""
+    tool = _make_tool(tmp_path)
+    result = await tool.execute(
+        action="add",
+        message="Shanghai reminder",
+        at="2026-03-18T14:00:00",
+        tz="Asia/Shanghai",
+    )
+    assert "Created job" in result
+
+    jobs = tool._cron.list_jobs()
+    assert len(jobs) == 1
+    # Asia/Shanghai is UTC+8, so 14:00 Shanghai = 06:00 UTC
+    expected_dt = datetime(2026, 3, 18, 14, 0, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
+    expected_ms = int(expected_dt.timestamp() * 1000)
+    assert jobs[0].schedule.at_ms == expected_ms
+
+
+@pytest.mark.asyncio
+async def test_at_with_tz_aware_datetime_preserves_original(tmp_path) -> None:
+    """Datetime that already has tzinfo should not be overridden by tz param."""
+    tool = _make_tool(tmp_path)
+    # Pass an aware datetime (UTC) with a different tz param
+    result = await tool.execute(
+        action="add",
+        message="UTC reminder",
+        at="2026-03-18T06:00:00+00:00",
+        tz="Asia/Shanghai",
+    )
+    assert "Created job" in result
+
+    jobs = tool._cron.list_jobs()
+    assert len(jobs) == 1
+    # The +00:00 offset should be preserved (dt.tzinfo is not None, so tz is ignored)
+    expected_dt = datetime(2026, 3, 18, 6, 0, 0, tzinfo=timezone.utc)
+    expected_ms = int(expected_dt.timestamp() * 1000)
+    assert jobs[0].schedule.at_ms == expected_ms
+
+
+@pytest.mark.asyncio
+async def test_tz_without_cron_or_at_fails(tmp_path) -> None:
+    """Passing tz without cron_expr or at should return an error."""
+    tool = _make_tool(tmp_path)
+    result = await tool.execute(
+        action="add",
+        message="Bad config",
+        tz="America/Vancouver",
+    )
+    assert "Error" in result
+    assert "tz can only be used with cron_expr or at" in result
+
+
+@pytest.mark.asyncio
+async def test_at_without_tz_unchanged(tmp_path) -> None:
+    """Naive datetime without tz should use system-local interpretation (existing behavior)."""
+    tool = _make_tool(tmp_path)
+    result = await tool.execute(
+        action="add",
+        message="Local reminder",
+        at="2026-03-18T14:00:00",
+    )
+    assert "Created job" in result
+
+    jobs = tool._cron.list_jobs()
+    assert len(jobs) == 1
+    # fromisoformat without tz → system local; just verify job was created
+    local_dt = datetime.fromisoformat("2026-03-18T14:00:00")
+    expected_ms = int(local_dt.timestamp() * 1000)
+    assert jobs[0].schedule.at_ms == expected_ms
+
+
+@pytest.mark.asyncio
+async def test_at_with_invalid_tz_fails(tmp_path) -> None:
+    """Invalid timezone should return an error."""
+    tool = _make_tool(tmp_path)
+    result = await tool.execute(
+        action="add",
+        message="Bad tz",
+        at="2026-03-18T14:00:00",
+        tz="Invalid/Timezone",
+    )
+    assert "Error" in result
+    assert "unknown timezone" in result

tests/test_mistral_provider.py

@@ -0,0 +1,22 @@
+"""Tests for the Mistral provider registration."""
+
+from nanobot.config.schema import ProvidersConfig
+from nanobot.providers.registry import PROVIDERS
+
+
+def test_mistral_config_field_exists():
+    """ProvidersConfig should have a mistral field."""
+    config = ProvidersConfig()
+    assert hasattr(config, "mistral")
+
+
+def test_mistral_provider_in_registry():
+    """Mistral should be registered in the provider registry."""
+    specs = {s.name: s for s in PROVIDERS}
+    assert "mistral" in specs
+
+    mistral = specs["mistral"]
+    assert mistral.env_key == "MISTRAL_API_KEY"
+    assert mistral.litellm_prefix == "mistral"
+    assert mistral.default_api_base == "https://api.mistral.ai/v1"
+    assert "mistral/" in mistral.skip_prefixes

tests/test_onboard_logic.py

@@ -0,0 +1,491 @@
+"""Unit tests for onboard core logic functions.
+
+These tests focus on the business logic behind the onboard wizard,
+without testing the interactive UI components.
+"""
+
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+
+from pydantic import BaseModel, Field
+
+from nanobot.cli import onboard_wizard
+
+# Import functions to test
+from nanobot.cli.commands import _merge_missing_defaults
+from nanobot.cli.onboard_wizard import (
+    _BACK_PRESSED,
+    _configure_pydantic_model,
+    _format_value,
+    _get_field_display_name,
+    _get_field_type_info,
+    run_onboard,
+)
+from nanobot.config.schema import Config
+from nanobot.utils.helpers import sync_workspace_templates
+
+
+class _SimpleDraftModel(BaseModel):
+    api_key: str = ""
+
+
+class _NestedDraftModel(BaseModel):
+    api_key: str = ""
+
+
+class _OuterDraftModel(BaseModel):
+    nested: _NestedDraftModel = Field(default_factory=_NestedDraftModel)
+
+
+class TestMergeMissingDefaults:
+    """Tests for _merge_missing_defaults recursive config merging."""
+
+    def test_adds_missing_top_level_keys(self):
+        existing = {"a": 1}
+        defaults = {"a": 1, "b": 2, "c": 3}
+
+        result = _merge_missing_defaults(existing, defaults)
+
+        assert result == {"a": 1, "b": 2, "c": 3}
+
+    def test_preserves_existing_values(self):
+        existing = {"a": "custom_value"}
+        defaults = {"a": "default_value"}
+
+        result = _merge_missing_defaults(existing, defaults)
+
+        assert result == {"a": "custom_value"}
+
+    def test_merges_nested_dicts_recursively(self):
+        existing = {
+            "level1": {
+                "level2": {
+                    "existing": "kept",
+                }
+            }
+        }
+        defaults = {
+            "level1": {
+                "level2": {
+                    "existing": "replaced",
+                    "added": "new",
+                },
+                "level2b": "also_new",
+            }
+        }
+
+        result = _merge_missing_defaults(existing, defaults)
+
+        assert result == {
+            "level1": {
+                "level2": {
+                    "existing": "kept",
+                    "added": "new",
+                },
+                "level2b": "also_new",
+            }
+        }
+
+    def test_returns_existing_if_not_dict(self):
+        assert _merge_missing_defaults("string", {"a": 1}) == "string"
+        assert _merge_missing_defaults([1, 2, 3], {"a": 1}) == [1, 2, 3]
+        assert _merge_missing_defaults(None, {"a": 1}) is None
+        assert _merge_missing_defaults(42, {"a": 1}) == 42
+
+    def test_returns_existing_if_defaults_not_dict(self):
+        assert _merge_missing_defaults({"a": 1}, "string") == {"a": 1}
+        assert _merge_missing_defaults({"a": 1}, None) == {"a": 1}
+
+    def test_handles_empty_dicts(self):
+        assert _merge_missing_defaults({}, {"a": 1}) == {"a": 1}
+        assert _merge_missing_defaults({"a": 1}, {}) == {"a": 1}
+        assert _merge_missing_defaults({}, {}) == {}
+
+    def test_backfills_channel_config(self):
+        """Real-world scenario: backfill missing channel fields."""
+        existing_channel = {
+            "enabled": False,
+            "appId": "",
+            "secret": "",
+        }
+        default_channel = {
+            "enabled": False,
+            "appId": "",
+            "secret": "",
+            "msgFormat": "plain",
+            "allowFrom": [],
+        }
+
+        result = _merge_missing_defaults(existing_channel, default_channel)
+
+        assert result["msgFormat"] == "plain"
+        assert result["allowFrom"] == []
+
+
+class TestGetFieldTypeInfo:
+    """Tests for _get_field_type_info type extraction."""
+
+    def test_extracts_str_type(self):
+        class Model(BaseModel):
+            field: str
+
+        type_name, inner = _get_field_type_info(Model.model_fields["field"])
+        assert type_name == "str"
+        assert inner is None
+
+    def test_extracts_int_type(self):
+        class Model(BaseModel):
+            count: int
+
+        type_name, inner = _get_field_type_info(Model.model_fields["count"])
+        assert type_name == "int"
+        assert inner is None
+
+    def test_extracts_bool_type(self):
+        class Model(BaseModel):
+            enabled: bool
+
+        type_name, inner = _get_field_type_info(Model.model_fields["enabled"])
+        assert type_name == "bool"
+        assert inner is None
+
+    def test_extracts_float_type(self):
+        class Model(BaseModel):
+            ratio: float
+
+        type_name, inner = _get_field_type_info(Model.model_fields["ratio"])
+        assert type_name == "float"
+        assert inner is None
+
+    def test_extracts_list_type_with_item_type(self):
+        class Model(BaseModel):
+            items: list[str]
+
+        type_name, inner = _get_field_type_info(Model.model_fields["items"])
+        assert type_name == "list"
+        assert inner is str
+
+    def test_extracts_list_type_without_item_type(self):
+        # Plain list without type param falls back to str
+        class Model(BaseModel):
+            items: list  # type: ignore
+
+        # Plain list annotation doesn't match list check, returns str
+        type_name, inner = _get_field_type_info(Model.model_fields["items"])
+        assert type_name == "str"  # Falls back to str for untyped list
+        assert inner is None
+
+    def test_extracts_dict_type(self):
+        # Plain dict without type param falls back to str
+        class Model(BaseModel):
+            data: dict  # type: ignore
+
+        # Plain dict annotation doesn't match dict check, returns str
+        type_name, inner = _get_field_type_info(Model.model_fields["data"])
+        assert type_name == "str"  # Falls back to str for untyped dict
+        assert inner is None
+
+    def test_extracts_optional_type(self):
+        class Model(BaseModel):
+            optional: str | None = None
+
+        type_name, inner = _get_field_type_info(Model.model_fields["optional"])
+        # Should unwrap Optional and get str
+        assert type_name == "str"
+        assert inner is None
+
+    def test_extracts_nested_model_type(self):
+        class Inner(BaseModel):
+            x: int
+
+        class Outer(BaseModel):
+            nested: Inner
+
+        type_name, inner = _get_field_type_info(Outer.model_fields["nested"])
+        assert type_name == "model"
+        assert inner is Inner
+
+    def test_handles_none_annotation(self):
+        """Field with None annotation defaults to str."""
+
+        class Model(BaseModel):
+            field: Any = None
+
+        # Create a mock field_info with None annotation
+        field_info = SimpleNamespace(annotation=None)
+        type_name, inner = _get_field_type_info(field_info)
+        assert type_name == "str"
+        assert inner is None
+
+
+class TestGetFieldDisplayName:
+    """Tests for _get_field_display_name human-readable name generation."""
+
+    def test_uses_description_if_present(self):
+        class Model(BaseModel):
+            api_key: str = Field(description="API Key for authentication")
+
+        name = _get_field_display_name("api_key", Model.model_fields["api_key"])
+        assert name == "API Key for authentication"
+
+    def test_converts_snake_case_to_title(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("user_name", field_info)
+        assert name == "User Name"
+
+    def test_adds_url_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("api_url", field_info)
+        # Title case: "Api Url"
+        assert "Url" in name and "Api" in name
+
+    def test_adds_path_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("file_path", field_info)
+        assert "Path" in name and "File" in name
+
+    def test_adds_id_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("user_id", field_info)
+        # Title case: "User Id"
+        assert "Id" in name and "User" in name
+
+    def test_adds_key_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("api_key", field_info)
+        assert "Key" in name and "Api" in name
+
+    def test_adds_token_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("auth_token", field_info)
+        assert "Token" in name and "Auth" in name
+
+    def test_adds_seconds_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("timeout_s", field_info)
+        # Contains "(Seconds)" with title case
+        assert "(Seconds)" in name or "(seconds)" in name
+
+    def test_adds_ms_suffix(self):
+        field_info = SimpleNamespace(description=None)
+        name = _get_field_display_name("delay_ms", field_info)
+        # Contains "(Ms)" or "(ms)"
+        assert "(Ms)" in name or "(ms)" in name
+
+
+class TestFormatValue:
+    """Tests for _format_value display formatting."""
+
+    def test_formats_none_as_not_set(self):
+        assert "not set" in _format_value(None)
+
+    def test_formats_empty_string_as_not_set(self):
+        assert "not set" in _format_value("")
+
+    def test_formats_empty_dict_as_not_set(self):
+        assert "not set" in _format_value({})
+
+    def test_formats_empty_list_as_not_set(self):
+        assert "not set" in _format_value([])
+
+    def test_formats_string_value(self):
+        result = _format_value("hello")
+        assert "hello" in result
+
+    def test_formats_list_value(self):
+        result = _format_value(["a", "b"])
+        assert "a" in result or "b" in result
+
+    def test_formats_dict_value(self):
+        result = _format_value({"key": "value"})
+        assert "key" in result or "value" in result
+
+    def test_formats_int_value(self):
+        result = _format_value(42)
+        assert "42" in result
+
+    def test_formats_bool_true(self):
+        result = _format_value(True)
+        assert "true" in result.lower() or "✓" in result
+
+    def test_formats_bool_false(self):
+        result = _format_value(False)
+        assert "false" in result.lower() or "✗" in result
+
+
+class TestSyncWorkspaceTemplates:
+    """Tests for sync_workspace_templates file synchronization."""
+
+    def test_creates_missing_files(self, tmp_path):
+        """Should create template files that don't exist."""
+        workspace = tmp_path / "workspace"
+
+        added = sync_workspace_templates(workspace, silent=True)
+
+        # Check that some files were created
+        assert isinstance(added, list)
+        # The actual files depend on the templates directory
+
+    def test_does_not_overwrite_existing_files(self, tmp_path):
+        """Should not overwrite files that already exist."""
+        workspace = tmp_path / "workspace"
+        workspace.mkdir(parents=True)
+        (workspace / "AGENTS.md").write_text("existing content")
+
+        sync_workspace_templates(workspace, silent=True)
+
+        # Existing file should not be changed
+        content = (workspace / "AGENTS.md").read_text()
+        assert content == "existing content"
+
+    def test_creates_memory_directory(self, tmp_path):
+        """Should create memory directory structure."""
+        workspace = tmp_path / "workspace"
+
+        sync_workspace_templates(workspace, silent=True)
+
+        assert (workspace / "memory").exists() or (workspace / "skills").exists()
+
+    def test_returns_list_of_added_files(self, tmp_path):
+        """Should return list of relative paths for added files."""
+        workspace = tmp_path / "workspace"
+
+        added = sync_workspace_templates(workspace, silent=True)
+
+        assert isinstance(added, list)
+        # All paths should be relative to workspace
+        for path in added:
+            assert not Path(path).is_absolute()
+
+
+class TestProviderChannelInfo:
+    """Tests for provider and channel info retrieval."""
+
+    def test_get_provider_names_returns_dict(self):
+        from nanobot.cli.onboard_wizard import _get_provider_names
+
+        names = _get_provider_names()
+        assert isinstance(names, dict)
+        assert len(names) > 0
+        # Should include common providers
+        assert "openai" in names or "anthropic" in names
+
+    def test_get_channel_names_returns_dict(self):
+        from nanobot.cli.onboard_wizard import _get_channel_names
+
+        names = _get_channel_names()
+        assert isinstance(names, dict)
+        # Should include at least some channels
+        assert len(names) >= 0
+
+    def test_get_provider_info_returns_valid_structure(self):
+        from nanobot.cli.onboard_wizard import _get_provider_info
+
+        info = _get_provider_info()
+        assert isinstance(info, dict)
+        # Each value should be a tuple with expected structure
+        for provider_name, value in info.items():
+            assert isinstance(value, tuple)
+            assert len(value) == 4  # (display_name, needs_api_key, needs_api_base, env_var)
+
+
+class TestConfigurePydanticModelDrafts:
+    @staticmethod
+    def _patch_prompt_helpers(monkeypatch, tokens, text_value="secret"):
+        sequence = iter(tokens)
+
+        def fake_select(_prompt, choices, default=None):
+            token = next(sequence)
+            if token == "first":
+                return choices[0]
+            if token == "done":
+                return "✓ Done"
+            if token == "back":
+                return _BACK_PRESSED
+            return token
+
+        monkeypatch.setattr(onboard_wizard, "_select_with_back", fake_select)
+        monkeypatch.setattr(onboard_wizard, "_show_config_panel", lambda *_args, **_kwargs: None)
+        monkeypatch.setattr(
+            onboard_wizard, "_input_with_existing", lambda *_args, **_kwargs: text_value
+        )
+
+    def test_discarding_section_keeps_original_model_unchanged(self, monkeypatch):
+        model = _SimpleDraftModel()
+        self._patch_prompt_helpers(monkeypatch, ["first", "back"])
+
+        result = _configure_pydantic_model(model, "Simple")
+
+        assert result is None
+        assert model.api_key == ""
+
+    def test_completing_section_returns_updated_draft(self, monkeypatch):
+        model = _SimpleDraftModel()
+        self._patch_prompt_helpers(monkeypatch, ["first", "done"])
+
+        result = _configure_pydantic_model(model, "Simple")
+
+        assert result is not None
+        updated = cast(_SimpleDraftModel, result)
+        assert updated.api_key == "secret"
+        assert model.api_key == ""
+
+    def test_nested_section_back_discards_nested_edits(self, monkeypatch):
+        model = _OuterDraftModel()
+        self._patch_prompt_helpers(monkeypatch, ["first", "first", "back", "done"])
+
+        result = _configure_pydantic_model(model, "Outer")
+
+        assert result is not None
+        updated = cast(_OuterDraftModel, result)
+        assert updated.nested.api_key == ""
+        assert model.nested.api_key == ""
+
+    def test_nested_section_done_commits_nested_edits(self, monkeypatch):
+        model = _OuterDraftModel()
+        self._patch_prompt_helpers(monkeypatch, ["first", "first", "done", "done"])
+
+        result = _configure_pydantic_model(model, "Outer")
+
+        assert result is not None
+        updated = cast(_OuterDraftModel, result)
+        assert updated.nested.api_key == "secret"
+        assert model.nested.api_key == ""
+
+
+class TestRunOnboardExitBehavior:
+    def test_main_menu_interrupt_can_discard_unsaved_session_changes(self, monkeypatch):
+        initial_config = Config()
+
+        responses = iter(
+            [
+                "🤖 Configure Agent Settings",
+                KeyboardInterrupt(),
+                "🗑️ Exit Without Saving",
+            ]
+        )
+
+        class FakePrompt:
+            def __init__(self, response):
+                self.response = response
+
+            def ask(self):
+                if isinstance(self.response, BaseException):
+                    raise self.response
+                return self.response
+
+        def fake_select(*_args, **_kwargs):
+            return FakePrompt(next(responses))
+
+        def fake_configure_agents(config):
+            config.agents.defaults.model = "test/provider-model"
+
+        monkeypatch.setattr(onboard_wizard, "_show_main_menu_header", lambda: None)
+        monkeypatch.setattr(onboard_wizard.questionary, "select", fake_select)
+        monkeypatch.setattr(onboard_wizard, "_configure_agents", fake_configure_agents)
+
+        result = run_onboard(initial_config=initial_config)
+
+        assert result.should_save is False
+        assert result.config.model_dump(by_alias=True) == initial_config.model_dump(by_alias=True)

tests/test_qq_channel.py

@@ -1,11 +1,12 @@
+import tempfile
+from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
 
 from nanobot.bus.events import OutboundMessage
 from nanobot.bus.queue import MessageBus
-from nanobot.channels.qq import QQChannel
+from nanobot.channels.qq import QQChannel, QQConfig
-from nanobot.channels.qq import QQConfig
 
 
 class _FakeApi:
@@ -34,6 +35,7 @@ async def test_on_group_message_routes_to_group_chat_id() -> None:
         content="hello",
         group_openid="group123",
         author=SimpleNamespace(member_openid="user1"),
+        attachments=[],
     )
 
     await channel._on_message(data, is_group=True)
@@ -123,3 +125,38 @@ async def test_send_group_message_uses_markdown_when_configured() -> None:
         "msg_id": "msg1",
         "msg_seq": 2,
     }
+
+
+@pytest.mark.asyncio
+async def test_read_media_bytes_local_path() -> None:
+    channel = QQChannel(QQConfig(app_id="app", secret="secret"), MessageBus())
+
+    with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as f:
+        f.write(b"\x89PNG\r\n")
+        tmp_path = f.name
+
+    data, filename = await channel._read_media_bytes(tmp_path)
+    assert data == b"\x89PNG\r\n"
+    assert filename == Path(tmp_path).name
+
+
+@pytest.mark.asyncio
+async def test_read_media_bytes_file_uri() -> None:
+    channel = QQChannel(QQConfig(app_id="app", secret="secret"), MessageBus())
+
+    with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
+        f.write(b"JFIF")
+        tmp_path = f.name
+
+    data, filename = await channel._read_media_bytes(f"file://{tmp_path}")
+    assert data == b"JFIF"
+    assert filename == Path(tmp_path).name
+
+
+@pytest.mark.asyncio
+async def test_read_media_bytes_missing_file() -> None:
+    channel = QQChannel(QQConfig(app_id="app", secret="secret"), MessageBus())
+
+    data, filename = await channel._read_media_bytes("/nonexistent/path/image.png")
+    assert data is None
+    assert filename is None