Compare commits

...

4 Commits

Author SHA1 Message Date
Hua
0859d5c9f6 Merge remote-tracking branch 'origin/main'
Some checks failed
Test Suite / test (3.11) (push) Failing after 1m3s
Test Suite / test (3.12) (push) Failing after 1m5s
Test Suite / test (3.13) (push) Failing after 1m2s
# Conflicts:
#	nanobot/channels/telegram.py
2026-03-19 16:47:40 +08:00
Hua
395fdc16f9 feat(qq): serve public media via gateway 2026-03-19 16:27:29 +08:00
Xubin Ren
dd7e3e499f fix: separate Telegram connection pools and add timeout retry to prevent pool exhaustion
The root cause of "Pool timeout" errors is that long-polling (getUpdates)
and outbound API calls (send_message, send_photo, etc.) shared the same
HTTPXRequest pool — polling holds connections indefinitely, starving sends
under concurrent load (e.g. cron jobs + user chat).

- Split into two independent pools: API calls (default 32) and polling (4)
- Expose connection_pool_size / pool_timeout in TelegramConfig for tuning
- Add _call_with_retry() with exponential backoff (3 attempts) on TimedOut
- Apply retry to _send_text and remote media URL sends
2026-03-19 16:15:41 +08:00
mamamiyear
d9cb729596 feat: support feishu code block 2026-03-19 13:59:31 +08:00
12 changed files with 915 additions and 35 deletions

View File

@@ -33,6 +33,7 @@ Do not commit real API keys, tokens, chat logs, or workspace data. Keep local se
- `/skill` currently supports `search`, `install`, `uninstall`, `list`, and `update`. Keep subcommand dispatch in `nanobot/agent/loop.py`. - `/skill` currently supports `search`, `install`, `uninstall`, `list`, and `update`. Keep subcommand dispatch in `nanobot/agent/loop.py`.
- `/mcp` supports the default `list` behavior (and explicit `/mcp list`) to show configured MCP servers and registered MCP tools. - `/mcp` supports the default `list` behavior (and explicit `/mcp list`) to show configured MCP servers and registered MCP tools.
- Agent runtime config should be hot-reloaded from the active `config.json` for safe in-process fields such as `tools.mcpServers`, `tools.web.*`, `tools.exec.*`, `tools.restrictToWorkspace`, `agents.defaults.model`, `agents.defaults.maxToolIterations`, `agents.defaults.contextWindowTokens`, `agents.defaults.maxTokens`, `agents.defaults.temperature`, `agents.defaults.reasoningEffort`, `channels.sendProgress`, and `channels.sendToolHints`. Channel connection settings and provider credentials still require a restart. - Agent runtime config should be hot-reloaded from the active `config.json` for safe in-process fields such as `tools.mcpServers`, `tools.web.*`, `tools.exec.*`, `tools.restrictToWorkspace`, `agents.defaults.model`, `agents.defaults.maxToolIterations`, `agents.defaults.contextWindowTokens`, `agents.defaults.maxTokens`, `agents.defaults.temperature`, `agents.defaults.reasoningEffort`, `channels.sendProgress`, and `channels.sendToolHints`. Channel connection settings and provider credentials still require a restart.
- QQ outbound media uses QQ's URL-based rich-media API. Remote `http(s)` image URLs can be sent directly. Local files are allowed from two controlled locations only: the configured `mediaPublicDir` inside `workspace/public`, and generated image files under `workspace/out`, which the QQ channel may hard-link into `public/` automatically before sending. Do not auto-publish from any other directory.
- `/skill` shells out to `npx clawhub@latest`; it requires Node.js/`npx` at runtime. - `/skill` shells out to `npx clawhub@latest`; it requires Node.js/`npx` at runtime.
- `/skill uninstall` runs in a non-interactive context, so keep passing `--yes` when shelling out to ClawHub. - `/skill uninstall` runs in a non-interactive context, so keep passing `--yes` when shelling out to ClawHub.
- Treat empty `/skill search` output as a user-visible "no results" case rather than a silent success. Surface npm/registry failures directly to the user. - Treat empty `/skill search` output as a user-visible "no results" case rather than a silent success. Surface npm/registry failures directly to the user.

View File

@@ -699,12 +699,22 @@ Uses **botpy SDK** with WebSocket — no public IP required. Currently supports
"enabled": true, "enabled": true,
"appId": "YOUR_APP_ID", "appId": "YOUR_APP_ID",
"secret": "YOUR_APP_SECRET", "secret": "YOUR_APP_SECRET",
"allowFrom": ["YOUR_OPENID"] "allowFrom": ["YOUR_OPENID"],
"mediaBaseUrl": "https://bot.example.com/public/qq/",
"mediaPublicDir": "public/qq",
"mediaTtlSeconds": 600
} }
} }
} }
``` ```
`mediaBaseUrl` is optional, but it is required if you want nanobot to send local screenshots or
other local image files through QQ. `mediaPublicDir` is resolved against the active startup
workspace and must stay under `workspace/public`, because the built-in gateway HTTP server only
serves that tree at `/public/`. nanobot accepts local QQ media from two places only: files already
under `mediaPublicDir`, and generated image files under `workspace/out`, which nanobot will
hard-link into `mediaPublicDir` automatically before sending.
Multi-bot example: Multi-bot example:
```json ```json
@@ -739,6 +749,16 @@ nanobot gateway
Now send a message to the bot from QQ — it should respond! Now send a message to the bot from QQ — it should respond!
Outbound QQ media always uses the QQ `url`-based rich-media API. Remote `http(s)` image URLs can be
sent directly. Local image files can also be sent when `mediaBaseUrl` points to a public URL and
`mediaPublicDir` matches a directory under `workspace/public`; nanobot maps that local public path
to a URL and then sends that URL through QQ. The built-in gateway route exposes
`workspace/public` as `/public/`, so a common setup is `mediaBaseUrl = https://your-host/public/qq/`.
If you generate screenshots under `workspace/out`, nanobot will automatically create a hard link in
`workspace/public/qq` first, then send that public URL. Files outside `mediaPublicDir` and
`workspace/out` are rejected. Without that publishing config, local files still fall back to a text
notice.
</details> </details>
<details> <details>

View File

@@ -189,6 +189,10 @@ def _extract_post_content(content_json: dict) -> tuple[str, list[str]]:
texts.append(el.get("text", "")) texts.append(el.get("text", ""))
elif tag == "at": elif tag == "at":
texts.append(f"@{el.get('user_name', 'user')}") texts.append(f"@{el.get('user_name', 'user')}")
elif tag == "code_block":
lang = el.get("language", "")
code_text = el.get("text", "")
texts.append(f"\n```{lang}\n{code_text}\n```\n")
elif tag == "img" and (key := el.get("image_key")): elif tag == "img" and (key := el.get("image_key")):
images.append(key) images.append(key)
return (" ".join(texts).strip() or None), images return (" ".join(texts).strip() or None), images
@@ -1014,7 +1018,7 @@ class FeishuChannel(BaseChannel):
event = data.event event = data.event
message = event.message message = event.message
sender = event.sender sender = event.sender
# Deduplication check # Deduplication check
message_id = message.message_id message_id = message.message_id
if message_id in self._processed_message_ids: if message_id in self._processed_message_ids:

View File

@@ -1,8 +1,12 @@
"""QQ channel implementation using botpy SDK.""" """QQ channel implementation using botpy SDK."""
import asyncio import asyncio
import os
import secrets
from collections import deque from collections import deque
from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from urllib.parse import quote, urljoin
from loguru import logger from loguru import logger
@@ -10,6 +14,8 @@ from nanobot.bus.events import OutboundMessage
from nanobot.bus.queue import MessageBus from nanobot.bus.queue import MessageBus
from nanobot.channels.base import BaseChannel from nanobot.channels.base import BaseChannel
from nanobot.config.schema import QQConfig, QQInstanceConfig from nanobot.config.schema import QQConfig, QQInstanceConfig
from nanobot.security.network import validate_url_target
from nanobot.utils.helpers import detect_image_mime, ensure_dir
try: try:
import botpy import botpy
@@ -60,13 +66,225 @@ class QQChannel(BaseChannel):
def default_config(cls) -> dict[str, object]: def default_config(cls) -> dict[str, object]:
return QQConfig().model_dump(by_alias=True) return QQConfig().model_dump(by_alias=True)
def __init__(self, config: QQConfig | QQInstanceConfig, bus: MessageBus): def __init__(
self,
config: QQConfig | QQInstanceConfig,
bus: MessageBus,
workspace: str | Path | None = None,
):
super().__init__(config, bus) super().__init__(config, bus)
self.config: QQConfig | QQInstanceConfig = config self.config: QQConfig | QQInstanceConfig = config
self._client: "botpy.Client | None" = None self._client: "botpy.Client | None" = None
self._processed_ids: deque = deque(maxlen=1000) self._processed_ids: deque = deque(maxlen=1000)
self._msg_seq: int = 1 # 消息序列号,避免被 QQ API 去重 self._msg_seq: int = 1 # 消息序列号,避免被 QQ API 去重
self._chat_type_cache: dict[str, str] = {} self._chat_type_cache: dict[str, str] = {}
self._workspace = Path(workspace).expanduser() if workspace is not None else None
self._cleanup_tasks: set[asyncio.Task[None]] = set()
@staticmethod
def _is_remote_media(path: str) -> bool:
"""Return True when the outbound media reference is a remote URL."""
return path.startswith(("http://", "https://"))
@staticmethod
def _failed_media_notice(path: str, reason: str | None = None) -> str:
"""Render a user-visible fallback notice for unsent QQ media."""
name = Path(path).name or path
return f"[Failed to send: {name}{f' - {reason}' if reason else ''}]"
def _workspace_root(self) -> Path:
"""Return the active workspace root used by QQ publishing."""
return (self._workspace or Path.cwd()).resolve(strict=False)
def _public_root(self) -> Path:
"""Return the fixed public tree served by the gateway HTTP route."""
return ensure_dir(self._workspace_root() / "public")
def _out_root(self) -> Path:
"""Return the default workspace out directory used for generated artifacts."""
return self._workspace_root() / "out"
def _resolve_media_public_dir(self) -> tuple[Path | None, str | None]:
"""Resolve the local publish directory for QQ media under workspace/public."""
configured = Path(self.config.media_public_dir).expanduser()
if configured.is_absolute():
resolved = configured.resolve(strict=False)
else:
resolved = (self._workspace_root() / configured).resolve(strict=False)
public_root = self._public_root()
try:
resolved.relative_to(public_root)
except ValueError:
return None, f"QQ mediaPublicDir must stay under {public_root}"
return ensure_dir(resolved), None
@staticmethod
def _guess_image_suffix(path: Path, mime_type: str | None) -> str:
"""Pick a reasonable output suffix for published QQ images."""
if path.suffix:
return path.suffix.lower()
return {
"image/png": ".png",
"image/jpeg": ".jpg",
"image/gif": ".gif",
"image/webp": ".webp",
}.get(mime_type or "", ".bin")
@staticmethod
def _is_image_file(path: Path) -> bool:
"""Validate that a local file looks like an image supported by QQ rich media."""
try:
with path.open("rb") as f:
header = f.read(16)
except OSError:
return False
return detect_image_mime(header) is not None
@staticmethod
def _detect_image_mime(path: Path) -> str | None:
"""Detect image mime type from the leading bytes of a file."""
try:
with path.open("rb") as f:
return detect_image_mime(f.read(16))
except OSError:
return None
async def _delete_published_media_later(self, path: Path, delay_seconds: int) -> None:
"""Delete an auto-published QQ media file after a grace period."""
try:
await asyncio.sleep(delay_seconds)
path.unlink(missing_ok=True)
except Exception as e:
logger.debug("Failed to delete published QQ media {}: {}", path, e)
def _schedule_media_cleanup(self, path: Path) -> None:
"""Best-effort cleanup for auto-published local QQ media."""
if self.config.media_ttl_seconds <= 0:
return
task = asyncio.create_task(
self._delete_published_media_later(path, self.config.media_ttl_seconds)
)
self._cleanup_tasks.add(task)
task.add_done_callback(self._cleanup_tasks.discard)
def _try_link_out_media_into_public(
self,
source: Path,
public_dir: Path,
) -> tuple[Path | None, str | None]:
"""Hard-link a generated workspace/out media file into public/qq."""
out_root = self._out_root().resolve(strict=False)
try:
source.relative_to(out_root)
except ValueError:
return None, f"QQ local media must stay under {public_dir} or {out_root}"
if not self._is_image_file(source):
return None, "QQ local media must be an image"
mime_type = self._detect_image_mime(source)
suffix = self._guess_image_suffix(source, mime_type)
published = public_dir / f"{source.stem}-{secrets.token_urlsafe(6)}{suffix}"
try:
os.link(source, published)
except OSError as e:
logger.warning("Failed to hard-link QQ media {} -> {}: {}", source, published, e)
return None, "failed to publish local file"
self._schedule_media_cleanup(published)
return published, None
async def _publish_local_media(self, media_path: str) -> tuple[str | None, str | None]:
"""Map a local public QQ media file, or a generated out file, to its served URL."""
if not self.config.media_base_url:
return None, "QQ local media publishing is not configured"
source = Path(media_path).expanduser()
try:
resolved = source.resolve(strict=True)
except FileNotFoundError:
return None, "local file not found"
except OSError as e:
logger.warning("Failed to resolve QQ media path {}: {}", media_path, e)
return None, "local file unavailable"
if not resolved.is_file():
return None, "local file not found"
public_dir, dir_error = self._resolve_media_public_dir()
if public_dir is None:
return None, dir_error
try:
relative_path = resolved.relative_to(public_dir)
except ValueError:
published, publish_error = self._try_link_out_media_into_public(resolved, public_dir)
if published is None:
return None, publish_error
relative_path = published.relative_to(public_dir)
media_url = urljoin(
f"{self.config.media_base_url.rstrip('/')}/",
quote(relative_path.as_posix(), safe="/"),
)
return media_url, None
def _next_msg_seq(self) -> int:
"""Return the next QQ message sequence number."""
self._msg_seq += 1
return self._msg_seq
async def _post_text_message(self, chat_id: str, msg_type: str, content: str, msg_id: str | None) -> None:
"""Send a plain-text QQ message."""
payload = {
"msg_type": 0,
"content": content,
"msg_id": msg_id,
"msg_seq": self._next_msg_seq(),
}
if msg_type == "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 _post_remote_media_message(
self,
chat_id: str,
msg_type: str,
media_url: str,
content: str | None,
msg_id: str | None,
) -> None:
"""Send one QQ remote image URL as a rich-media message."""
if msg_type == "group":
media = await self._client.api.post_group_file(
group_openid=chat_id,
file_type=1,
url=media_url,
srv_send_msg=False,
)
await self._client.api.post_group_message(
group_openid=chat_id,
msg_type=7,
content=content,
media=media,
msg_id=msg_id,
msg_seq=self._next_msg_seq(),
)
else:
media = await self._client.api.post_c2c_file(
openid=chat_id,
file_type=1,
url=media_url,
srv_send_msg=False,
)
await self._client.api.post_c2c_message(
openid=chat_id,
msg_type=7,
content=content,
media=media,
msg_id=msg_id,
msg_seq=self._next_msg_seq(),
)
async def start(self) -> None: async def start(self) -> None:
"""Start the QQ bot.""" """Start the QQ bot."""
@@ -98,6 +316,9 @@ class QQChannel(BaseChannel):
async def stop(self) -> None: async def stop(self) -> None:
"""Stop the QQ bot.""" """Stop the QQ bot."""
self._running = False self._running = False
for task in list(self._cleanup_tasks):
task.cancel()
self._cleanup_tasks.clear()
if self._client: if self._client:
try: try:
await self._client.close() await self._client.close()
@@ -113,24 +334,53 @@ class QQChannel(BaseChannel):
try: try:
msg_id = msg.metadata.get("message_id") msg_id = msg.metadata.get("message_id")
self._msg_seq += 1
msg_type = self._chat_type_cache.get(msg.chat_id, "c2c") msg_type = self._chat_type_cache.get(msg.chat_id, "c2c")
if msg_type == "group": content_sent = False
await self._client.api.post_group_message( fallback_lines: list[str] = []
group_openid=msg.chat_id,
msg_type=0, for media_path in msg.media:
content=msg.content, resolved_media = media_path
msg_id=msg_id, if not self._is_remote_media(media_path):
msg_seq=self._msg_seq, resolved_media, publish_error = await self._publish_local_media(media_path)
) if not resolved_media:
else: logger.warning(
await self._client.api.post_c2c_message( "QQ outbound local media could not be published: {} ({})",
openid=msg.chat_id, media_path,
msg_type=0, publish_error,
content=msg.content, )
msg_id=msg_id, fallback_lines.append(
msg_seq=self._msg_seq, self._failed_media_notice(media_path, publish_error)
) )
continue
ok, error = validate_url_target(resolved_media)
if not ok:
logger.warning("QQ outbound media blocked by URL validation: {}", error)
fallback_lines.append(self._failed_media_notice(media_path, error))
continue
try:
await self._post_remote_media_message(
msg.chat_id,
msg_type,
resolved_media,
msg.content if msg.content and not content_sent else None,
msg_id,
)
if msg.content and not content_sent:
content_sent = True
except Exception as media_error:
logger.error("Error sending QQ media {}: {}", resolved_media, media_error)
fallback_lines.append(self._failed_media_notice(media_path))
text_parts: list[str] = []
if msg.content and not content_sent:
text_parts.append(msg.content)
if fallback_lines:
text_parts.extend(fallback_lines)
if text_parts:
await self._post_text_message(msg.chat_id, msg_type, "\n".join(text_parts), msg_id)
except Exception as e: except Exception as e:
logger.error("Error sending QQ message: {}", e) logger.error("Error sending QQ message: {}", e)

View File

@@ -9,6 +9,7 @@ import unicodedata
from loguru import logger from loguru import logger
from telegram import BotCommand, ReplyParameters, Update from telegram import BotCommand, ReplyParameters, Update
from telegram.error import TimedOut
from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters
from telegram.request import HTTPXRequest from telegram.request import HTTPXRequest
@@ -154,7 +155,8 @@ def _markdown_to_telegram_html(text: str) -> str:
return text return text
_SEND_MAX_RETRIES = 3
_SEND_RETRY_BASE_DELAY = 0.5 # seconds, doubled each retry
class TelegramChannel(BaseChannel): class TelegramChannel(BaseChannel):
""" """
Telegram channel using long polling. Telegram channel using long polling.
@@ -221,15 +223,29 @@ class TelegramChannel(BaseChannel):
self._running = True self._running = True
# Build the application with larger connection pool to avoid pool-timeout on long runs proxy = self.config.proxy or None
req = HTTPXRequest(
connection_pool_size=16, # Separate pools so long-polling (getUpdates) never starves outbound sends.
pool_timeout=5.0, api_request = HTTPXRequest(
connection_pool_size=self.config.connection_pool_size,
pool_timeout=self.config.pool_timeout,
connect_timeout=30.0, connect_timeout=30.0,
read_timeout=30.0, read_timeout=30.0,
proxy=self.config.proxy if self.config.proxy else None, proxy=proxy,
)
poll_request = HTTPXRequest(
connection_pool_size=4,
pool_timeout=self.config.pool_timeout,
connect_timeout=30.0,
read_timeout=30.0,
proxy=proxy,
)
builder = (
Application.builder()
.token(self.config.token)
.request(api_request)
.get_updates_request(poll_request)
) )
builder = Application.builder().token(self.config.token).request(req).get_updates_request(req)
self._app = builder.build() self._app = builder.build()
self._app.add_error_handler(self._on_error) self._app.add_error_handler(self._on_error)
@@ -365,7 +381,8 @@ class TelegramChannel(BaseChannel):
ok, error = validate_url_target(media_path) ok, error = validate_url_target(media_path)
if not ok: if not ok:
raise ValueError(f"unsafe media URL: {error}") raise ValueError(f"unsafe media URL: {error}")
await sender( await self._call_with_retry(
sender,
chat_id=chat_id, chat_id=chat_id,
**{param: media_path}, **{param: media_path},
reply_parameters=reply_params, reply_parameters=reply_params,
@@ -401,6 +418,21 @@ class TelegramChannel(BaseChannel):
else: else:
await self._send_text(chat_id, chunk, reply_params, thread_kwargs) await self._send_text(chat_id, chunk, reply_params, thread_kwargs)
async def _call_with_retry(self, fn, *args, **kwargs):
"""Call an async Telegram API function with retry on pool/network timeout."""
for attempt in range(1, _SEND_MAX_RETRIES + 1):
try:
return await fn(*args, **kwargs)
except TimedOut:
if attempt == _SEND_MAX_RETRIES:
raise
delay = _SEND_RETRY_BASE_DELAY * (2 ** (attempt - 1))
logger.warning(
"Telegram timeout (attempt {}/{}), retrying in {:.1f}s",
attempt, _SEND_MAX_RETRIES, delay,
)
await asyncio.sleep(delay)
async def _send_text( async def _send_text(
self, self,
chat_id: int, chat_id: int,
@@ -411,7 +443,8 @@ class TelegramChannel(BaseChannel):
"""Send a plain text message with HTML fallback.""" """Send a plain text message with HTML fallback."""
try: try:
html = _markdown_to_telegram_html(text) html = _markdown_to_telegram_html(text)
await self._app.bot.send_message( await self._call_with_retry(
self._app.bot.send_message,
chat_id=chat_id, text=html, parse_mode="HTML", chat_id=chat_id, text=html, parse_mode="HTML",
reply_parameters=reply_params, reply_parameters=reply_params,
**(thread_kwargs or {}), **(thread_kwargs or {}),
@@ -419,7 +452,8 @@ class TelegramChannel(BaseChannel):
except Exception as e: except Exception as e:
logger.warning("HTML parse failed, falling back to plain text: {}", e) logger.warning("HTML parse failed, falling back to plain text: {}", e)
try: try:
await self._app.bot.send_message( await self._call_with_retry(
self._app.bot.send_message,
chat_id=chat_id, chat_id=chat_id,
text=text, text=text,
reply_parameters=reply_params, reply_parameters=reply_params,

View File

@@ -495,6 +495,7 @@ def gateway(
from nanobot.config.paths import get_cron_dir from nanobot.config.paths import get_cron_dir
from nanobot.cron.service import CronService from nanobot.cron.service import CronService
from nanobot.cron.types import CronJob from nanobot.cron.types import CronJob
from nanobot.gateway.http import GatewayHttpServer
from nanobot.heartbeat.service import HeartbeatService from nanobot.heartbeat.service import HeartbeatService
from nanobot.session.manager import SessionManager from nanobot.session.manager import SessionManager
@@ -581,6 +582,7 @@ def gateway(
# Create channel manager # Create channel manager
channels = ChannelManager(config, bus) channels = ChannelManager(config, bus)
http_server = GatewayHttpServer(config.workspace_path, config.gateway.host, port)
def _pick_heartbeat_target() -> tuple[str, str]: def _pick_heartbeat_target() -> tuple[str, str]:
"""Pick a routable channel/chat target for heartbeat-triggered messages.""" """Pick a routable channel/chat target for heartbeat-triggered messages."""
@@ -638,6 +640,10 @@ def gateway(
else: else:
console.print("[yellow]Warning: No channels enabled[/yellow]") console.print("[yellow]Warning: No channels enabled[/yellow]")
console.print(
f"[green]✓[/green] Public files: {http_server.public_dir} -> /public/"
)
cron_status = cron.status() cron_status = cron.status()
if cron_status["jobs"] > 0: if cron_status["jobs"] > 0:
console.print(f"[green]✓[/green] Cron: {cron_status['jobs']} scheduled jobs") console.print(f"[green]✓[/green] Cron: {cron_status['jobs']} scheduled jobs")
@@ -648,6 +654,7 @@ def gateway(
try: try:
await cron.start() await cron.start()
await heartbeat.start() await heartbeat.start()
await http_server.start()
await asyncio.gather( await asyncio.gather(
agent.run(), agent.run(),
channels.start_all(), channels.start_all(),
@@ -659,6 +666,7 @@ def gateway(
heartbeat.stop() heartbeat.stop()
cron.stop() cron.stop()
agent.stop() agent.stop()
await http_server.stop()
await channels.stop_all() await channels.stop_all()
asyncio.run(run()) asyncio.run(run())

View File

@@ -46,6 +46,8 @@ class TelegramConfig(Base):
) )
reply_to_message: bool = False # If true, bot replies quote the original message reply_to_message: bool = False # If true, bot replies quote the original message
group_policy: Literal["open", "mention"] = "mention" # "mention" responds when @mentioned or replied to, "open" responds to all group_policy: Literal["open", "mention"] = "mention" # "mention" responds when @mentioned or replied to, "open" responds to all
connection_pool_size: int = 32 # Outbound Telegram API HTTP pool size
pool_timeout: float = 5.0 # Shared HTTP pool timeout for bot sends and getUpdates
class TelegramInstanceConfig(TelegramConfig): class TelegramInstanceConfig(TelegramConfig):
@@ -315,6 +317,9 @@ class QQConfig(Base):
app_id: str = "" # 机器人 ID (AppID) from q.qq.com app_id: str = "" # 机器人 ID (AppID) from q.qq.com
secret: str = "" # 机器人密钥 (AppSecret) from q.qq.com secret: str = "" # 机器人密钥 (AppSecret) from q.qq.com
allow_from: list[str] = Field(default_factory=list) # Allowed user openids allow_from: list[str] = Field(default_factory=list) # Allowed user openids
media_base_url: str = "" # Public base URL used to expose local QQ media files
media_public_dir: str = "public/qq" # Must stay under the active workspace/public tree
media_ttl_seconds: int = 600 # Delete published local QQ media after N seconds; <=0 keeps files
class QQInstanceConfig(QQConfig): class QQInstanceConfig(QQConfig):

View File

@@ -0,0 +1 @@
"""Gateway HTTP helpers."""

65
nanobot/gateway/http.py Normal file
View File

@@ -0,0 +1,65 @@
"""Minimal HTTP server for workspace-scoped public files."""
from __future__ import annotations
from pathlib import Path
from aiohttp import web
from loguru import logger
from nanobot.utils.helpers import ensure_dir
def get_public_dir(workspace: Path) -> Path:
"""Return the fixed public directory served by the gateway."""
return ensure_dir(workspace / "public")
def create_http_app(workspace: Path) -> web.Application:
"""Create the gateway HTTP app serving workspace/public."""
public_dir = get_public_dir(workspace)
app = web.Application()
async def health(_request: web.Request) -> web.Response:
return web.json_response({"ok": True})
app.router.add_get("/healthz", health)
app.router.add_static("/public/", path=str(public_dir), follow_symlinks=False, show_index=False)
return app
class GatewayHttpServer:
"""Small aiohttp server exposing only workspace/public."""
def __init__(self, workspace: Path, host: str, port: int):
self.workspace = workspace
self.host = host
self.port = port
self._app = create_http_app(workspace)
self._runner: web.AppRunner | None = None
self._site: web.TCPSite | None = None
@property
def public_dir(self) -> Path:
"""Return the served public directory."""
return get_public_dir(self.workspace)
async def start(self) -> None:
"""Start serving the HTTP routes."""
self._runner = web.AppRunner(self._app, access_log=None)
await self._runner.setup()
self._site = web.TCPSite(self._runner, host=self.host, port=self.port)
await self._site.start()
logger.info(
"Gateway HTTP server listening on {}:{} (public dir: {})",
self.host,
self.port,
self.public_dir,
)
async def stop(self) -> None:
"""Stop the HTTP server."""
if self._runner:
await self._runner.cleanup()
self._runner = None
self._site = None

View File

@@ -0,0 +1,44 @@
import os
from pathlib import Path
import pytest
from aiohttp.test_utils import make_mocked_request
from nanobot.gateway.http import create_http_app, get_public_dir
@pytest.mark.asyncio
async def test_gateway_public_route_maps_requests_into_workspace_public(tmp_path) -> None:
public_dir = get_public_dir(tmp_path)
file_path = public_dir / "hello.txt"
file_path.write_text("hello", encoding="utf-8")
app = create_http_app(tmp_path)
request = make_mocked_request("GET", "/public/hello.txt", app=app)
match = await app.router.resolve(request)
assert match.route.resource.canonical == "/public"
assert match["filename"] == "hello.txt"
assert Path(getattr(match.route.resource, "_directory")) == public_dir
@pytest.mark.asyncio
async def test_gateway_public_route_disables_symlink_following_and_allows_hard_links(tmp_path) -> None:
out_dir = tmp_path / "out"
out_dir.mkdir()
source = out_dir / "shot.png"
source.write_bytes(b"png")
public_dir = get_public_dir(tmp_path) / "qq"
public_dir.mkdir()
published = public_dir / "shot.png"
os.link(source, published)
app = create_http_app(tmp_path)
request = make_mocked_request("GET", "/public/qq/shot.png", app=app)
match = await app.router.resolve(request)
assert os.stat(source).st_ino == os.stat(published).st_ino
assert match.route.resource.canonical == "/public"
assert match["filename"] == "qq/shot.png"
assert getattr(match.route.resource, "_follow_symlinks") is False

View File

@@ -1,3 +1,4 @@
import os
from types import SimpleNamespace from types import SimpleNamespace
import pytest import pytest
@@ -12,6 +13,8 @@ class _FakeApi:
def __init__(self) -> None: def __init__(self) -> None:
self.c2c_calls: list[dict] = [] self.c2c_calls: list[dict] = []
self.group_calls: list[dict] = [] self.group_calls: list[dict] = []
self.c2c_file_calls: list[dict] = []
self.group_file_calls: list[dict] = []
async def post_c2c_message(self, **kwargs) -> None: async def post_c2c_message(self, **kwargs) -> None:
self.c2c_calls.append(kwargs) self.c2c_calls.append(kwargs)
@@ -19,6 +22,14 @@ class _FakeApi:
async def post_group_message(self, **kwargs) -> None: async def post_group_message(self, **kwargs) -> None:
self.group_calls.append(kwargs) self.group_calls.append(kwargs)
async def post_c2c_file(self, **kwargs) -> dict:
self.c2c_file_calls.append(kwargs)
return {"file_info": "c2c-file-info", "file_uuid": "c2c-file", "ttl": 60}
async def post_group_file(self, **kwargs) -> dict:
self.group_file_calls.append(kwargs)
return {"file_info": "group-file-info", "file_uuid": "group-file", "ttl": 60}
class _FakeClient: class _FakeClient:
def __init__(self) -> None: def __init__(self) -> None:
@@ -94,3 +105,339 @@ async def test_send_c2c_message_uses_plain_text_c2c_api_with_msg_seq() -> None:
"msg_seq": 2, "msg_seq": 2,
} }
assert not channel._client.api.group_calls assert not channel._client.api.group_calls
@pytest.mark.asyncio
async def test_send_group_remote_media_url_uses_file_api_then_media_message(monkeypatch) -> None:
channel = QQChannel(QQConfig(app_id="app", secret="secret", allow_from=["*"]), MessageBus())
channel._client = _FakeClient()
channel._chat_type_cache["group123"] = "group"
monkeypatch.setattr("nanobot.channels.qq.validate_url_target", lambda url: (True, ""))
await channel.send(
OutboundMessage(
channel="qq",
chat_id="group123",
content="look",
media=["https://example.com/cat.jpg"],
metadata={"message_id": "msg1"},
)
)
assert channel._client.api.group_file_calls == [
{
"group_openid": "group123",
"file_type": 1,
"url": "https://example.com/cat.jpg",
"srv_send_msg": False,
}
]
assert channel._client.api.group_calls == [
{
"group_openid": "group123",
"msg_type": 7,
"content": "look",
"media": {"file_info": "group-file-info", "file_uuid": "group-file", "ttl": 60},
"msg_id": "msg1",
"msg_seq": 2,
}
]
assert channel._client.api.c2c_calls == []
@pytest.mark.asyncio
async def test_send_local_media_falls_back_to_text_notice_when_publishing_not_configured() -> None:
channel = QQChannel(QQConfig(app_id="app", secret="secret", allow_from=["*"]), MessageBus())
channel._client = _FakeClient()
await channel.send(
OutboundMessage(
channel="qq",
chat_id="user123",
content="hello",
media=["/tmp/demo.png"],
metadata={"message_id": "msg1"},
)
)
assert channel._client.api.c2c_file_calls == []
assert channel._client.api.group_file_calls == []
assert channel._client.api.c2c_calls == [
{
"openid": "user123",
"msg_type": 0,
"content": "hello\n[Failed to send: demo.png - QQ local media publishing is not configured]",
"msg_id": "msg1",
"msg_seq": 2,
}
]
@pytest.mark.asyncio
async def test_send_local_media_under_public_dir_uses_c2c_file_api(
monkeypatch,
tmp_path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
public_dir = workspace / "public" / "qq"
public_dir.mkdir(parents=True)
source = public_dir / "demo.png"
source.write_bytes(b"fake-png")
channel = QQChannel(
QQConfig(
app_id="app",
secret="secret",
allow_from=["*"],
media_base_url="https://files.example.com/public/qq",
media_public_dir="public/qq",
media_ttl_seconds=0,
),
MessageBus(),
workspace=workspace,
)
channel._client = _FakeClient()
monkeypatch.setattr("nanobot.channels.qq.validate_url_target", lambda url: (True, ""))
await channel.send(
OutboundMessage(
channel="qq",
chat_id="user123",
content="hello",
media=[str(source)],
metadata={"message_id": "msg1"},
)
)
assert channel._client.api.c2c_file_calls == [
{
"openid": "user123",
"file_type": 1,
"url": "https://files.example.com/public/qq/demo.png",
"srv_send_msg": False,
}
]
assert channel._client.api.c2c_calls == [
{
"openid": "user123",
"msg_type": 7,
"content": "hello",
"media": {"file_info": "c2c-file-info", "file_uuid": "c2c-file", "ttl": 60},
"msg_id": "msg1",
"msg_seq": 2,
}
]
@pytest.mark.asyncio
async def test_send_local_media_from_out_auto_links_into_public_then_uses_c2c_file_api(
monkeypatch,
tmp_path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
out_dir = workspace / "out"
out_dir.mkdir()
source = out_dir / "outside.png"
source.write_bytes(b"\x89PNG\r\n\x1a\nfake-png")
channel = QQChannel(
QQConfig(
app_id="app",
secret="secret",
allow_from=["*"],
media_base_url="https://files.example.com/public/qq",
media_public_dir="public/qq",
media_ttl_seconds=0,
),
MessageBus(),
workspace=workspace,
)
channel._client = _FakeClient()
monkeypatch.setattr("nanobot.channels.qq.validate_url_target", lambda url: (True, ""))
await channel.send(
OutboundMessage(
channel="qq",
chat_id="user123",
content="hello",
media=[str(source)],
metadata={"message_id": "msg1"},
)
)
published = list((workspace / "public" / "qq").iterdir())
assert len(published) == 1
assert published[0].name.startswith("outside-")
assert published[0].suffix == ".png"
assert os.stat(source).st_ino == os.stat(published[0]).st_ino
assert channel._client.api.c2c_file_calls == [
{
"openid": "user123",
"file_type": 1,
"url": f"https://files.example.com/public/qq/{published[0].name}",
"srv_send_msg": False,
}
]
assert channel._client.api.c2c_calls == [
{
"openid": "user123",
"msg_type": 7,
"content": "hello",
"media": {"file_info": "c2c-file-info", "file_uuid": "c2c-file", "ttl": 60},
"msg_id": "msg1",
"msg_seq": 2,
}
]
@pytest.mark.asyncio
async def test_send_local_media_outside_public_and_out_falls_back_to_text_notice(
monkeypatch,
tmp_path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
docs_dir = workspace / "docs"
docs_dir.mkdir()
source = docs_dir / "outside.png"
source.write_bytes(b"fake-png")
channel = QQChannel(
QQConfig(
app_id="app",
secret="secret",
allow_from=["*"],
media_base_url="https://files.example.com/public/qq",
media_public_dir="public/qq",
media_ttl_seconds=0,
),
MessageBus(),
workspace=workspace,
)
channel._client = _FakeClient()
monkeypatch.setattr("nanobot.channels.qq.validate_url_target", lambda url: (True, ""))
await channel.send(
OutboundMessage(
channel="qq",
chat_id="user123",
content="hello",
media=[str(source)],
metadata={"message_id": "msg1"},
)
)
assert channel._client.api.c2c_file_calls == []
assert channel._client.api.c2c_calls == [
{
"openid": "user123",
"msg_type": 0,
"content": "hello\n[Failed to send: outside.png - QQ local media must stay under "
f"{workspace / 'public' / 'qq'} or {workspace / 'out'}]",
"msg_id": "msg1",
"msg_seq": 2,
}
]
@pytest.mark.asyncio
async def test_send_local_media_symlink_to_outside_public_dir_is_rejected(
monkeypatch,
tmp_path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
public_dir = workspace / "public" / "qq"
public_dir.mkdir(parents=True)
outside = tmp_path / "secret.png"
outside.write_bytes(b"secret")
source = public_dir / "linked.png"
source.symlink_to(outside)
channel = QQChannel(
QQConfig(
app_id="app",
secret="secret",
allow_from=["*"],
media_base_url="https://files.example.com/public/qq",
media_public_dir="public/qq",
media_ttl_seconds=0,
),
MessageBus(),
workspace=workspace,
)
channel._client = _FakeClient()
monkeypatch.setattr("nanobot.channels.qq.validate_url_target", lambda url: (True, ""))
await channel.send(
OutboundMessage(
channel="qq",
chat_id="user123",
content="hello",
media=[str(source)],
metadata={"message_id": "msg1"},
)
)
assert channel._client.api.c2c_file_calls == []
assert channel._client.api.c2c_calls == [
{
"openid": "user123",
"msg_type": 0,
"content": "hello\n[Failed to send: linked.png - QQ local media must stay under "
f"{workspace / 'public' / 'qq'} or {workspace / 'out'}]",
"msg_id": "msg1",
"msg_seq": 2,
}
]
@pytest.mark.asyncio
async def test_send_non_image_media_from_out_falls_back_to_text_notice(
monkeypatch,
tmp_path,
) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
out_dir = workspace / "out"
out_dir.mkdir()
source = out_dir / "note.txt"
source.write_text("not an image", encoding="utf-8")
channel = QQChannel(
QQConfig(
app_id="app",
secret="secret",
allow_from=["*"],
media_base_url="https://files.example.com/public/qq",
media_public_dir="public/qq",
media_ttl_seconds=0,
),
MessageBus(),
workspace=workspace,
)
channel._client = _FakeClient()
monkeypatch.setattr("nanobot.channels.qq.validate_url_target", lambda url: (True, ""))
await channel.send(
OutboundMessage(
channel="qq",
chat_id="user123",
content="hello",
media=[str(source)],
metadata={"message_id": "msg1"},
)
)
assert channel._client.api.c2c_file_calls == []
assert channel._client.api.c2c_calls == [
{
"openid": "user123",
"msg_type": 0,
"content": "hello\n[Failed to send: note.txt - QQ local media must be an image]",
"msg_id": "msg1",
"msg_seq": 2,
}
]

View File

@@ -16,6 +16,10 @@ class _FakeHTTPXRequest:
self.kwargs = kwargs self.kwargs = kwargs
self.__class__.instances.append(self) self.__class__.instances.append(self)
@classmethod
def clear(cls) -> None:
cls.instances.clear()
class _FakeUpdater: class _FakeUpdater:
def __init__(self, on_start_polling) -> None: def __init__(self, on_start_polling) -> None:
@@ -142,7 +146,8 @@ def _make_telegram_update(
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_start_uses_request_proxy_without_builder_proxy(monkeypatch) -> None: async def test_start_creates_separate_pools_with_proxy(monkeypatch) -> None:
_FakeHTTPXRequest.clear()
config = TelegramConfig( config = TelegramConfig(
enabled=True, enabled=True,
token="123:abc", token="123:abc",
@@ -162,10 +167,106 @@ async def test_start_uses_request_proxy_without_builder_proxy(monkeypatch) -> No
await channel.start() await channel.start()
assert len(_FakeHTTPXRequest.instances) == 1 assert len(_FakeHTTPXRequest.instances) == 2
assert _FakeHTTPXRequest.instances[0].kwargs["proxy"] == config.proxy api_req, poll_req = _FakeHTTPXRequest.instances
assert builder.request_value is _FakeHTTPXRequest.instances[0] assert api_req.kwargs["proxy"] == config.proxy
assert builder.get_updates_request_value is _FakeHTTPXRequest.instances[0] assert poll_req.kwargs["proxy"] == config.proxy
assert api_req.kwargs["connection_pool_size"] == 32
assert poll_req.kwargs["connection_pool_size"] == 4
assert builder.request_value is api_req
assert builder.get_updates_request_value is poll_req
@pytest.mark.asyncio
async def test_start_respects_custom_pool_config(monkeypatch) -> None:
_FakeHTTPXRequest.clear()
config = TelegramConfig(
enabled=True,
token="123:abc",
allow_from=["*"],
connection_pool_size=32,
pool_timeout=10.0,
)
bus = MessageBus()
channel = TelegramChannel(config, bus)
app = _FakeApp(lambda: setattr(channel, "_running", False))
builder = _FakeBuilder(app)
monkeypatch.setattr("nanobot.channels.telegram.HTTPXRequest", _FakeHTTPXRequest)
monkeypatch.setattr(
"nanobot.channels.telegram.Application",
SimpleNamespace(builder=lambda: builder),
)
await channel.start()
api_req = _FakeHTTPXRequest.instances[0]
poll_req = _FakeHTTPXRequest.instances[1]
assert api_req.kwargs["connection_pool_size"] == 32
assert api_req.kwargs["pool_timeout"] == 10.0
assert poll_req.kwargs["pool_timeout"] == 10.0
@pytest.mark.asyncio
async def test_send_text_retries_on_timeout() -> None:
"""_send_text retries on TimedOut before succeeding."""
from telegram.error import TimedOut
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
call_count = 0
original_send = channel._app.bot.send_message
async def flaky_send(**kwargs):
nonlocal call_count
call_count += 1
if call_count <= 2:
raise TimedOut()
return await original_send(**kwargs)
channel._app.bot.send_message = flaky_send
import nanobot.channels.telegram as tg_mod
orig_delay = tg_mod._SEND_RETRY_BASE_DELAY
tg_mod._SEND_RETRY_BASE_DELAY = 0.01
try:
await channel._send_text(123, "hello", None, {})
finally:
tg_mod._SEND_RETRY_BASE_DELAY = orig_delay
assert call_count == 3
assert len(channel._app.bot.sent_messages) == 1
@pytest.mark.asyncio
async def test_send_text_gives_up_after_max_retries() -> None:
"""_send_text raises TimedOut after exhausting all retries."""
from telegram.error import TimedOut
channel = TelegramChannel(
TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
MessageBus(),
)
channel._app = _FakeApp(lambda: None)
async def always_timeout(**kwargs):
raise TimedOut()
channel._app.bot.send_message = always_timeout
import nanobot.channels.telegram as tg_mod
orig_delay = tg_mod._SEND_RETRY_BASE_DELAY
tg_mod._SEND_RETRY_BASE_DELAY = 0.01
try:
await channel._send_text(123, "hello", None, {})
finally:
tg_mod._SEND_RETRY_BASE_DELAY = orig_delay
assert channel._app.bot.sent_messages == []
def test_derive_topic_session_key_uses_thread_id() -> None: def test_derive_topic_session_key_uses_thread_id() -> None: