\ No newline at end of file
diff --git a/app-instance/backend-old/Dockerfile b/app-instance/backend-old/Dockerfile
deleted file mode 100644
index 5102b4f..0000000
--- a/app-instance/backend-old/Dockerfile
+++ /dev/null
@@ -1,43 +0,0 @@
-FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim
-
-# Install Node.js 20 for the WhatsApp bridge
-RUN apt-get update && \
- apt-get install -y --no-install-recommends curl ca-certificates gnupg git && \
- mkdir -p /etc/apt/keyrings && \
- curl -fsSL https://deb.nodesource.com/gpgkey/nodesource-repo.gpg.key | gpg --dearmor -o /etc/apt/keyrings/nodesource.gpg && \
- echo "deb [signed-by=/etc/apt/keyrings/nodesource.gpg] https://deb.nodesource.com/node_20.x nodistro main" > /etc/apt/sources.list.d/nodesource.list && \
- apt-get update && \
- apt-get install -y --no-install-recommends nodejs && \
- apt-get purge -y gnupg && \
- apt-get autoremove -y && \
- rm -rf /var/lib/apt/lists/*
-
-WORKDIR /app
-
-# Install Python dependencies first (cached layer)
-COPY pyproject.toml README.md LICENSE ./
-RUN mkdir -p nanobot bridge && touch nanobot/__init__.py && \
- uv pip install --system --no-cache . && \
- rm -rf nanobot bridge
-
-# Copy the full source and install
-COPY nanobot/ nanobot/
-COPY bridge/ bridge/
-COPY third_party/swarms/ third_party/swarms/
-RUN uv pip install --system --no-cache .
-
-# Build the WhatsApp bridge
-WORKDIR /app/bridge
-RUN git config --global url."https://github.com/".insteadOf "ssh://git@github.com/" && \
- git config --global url."https://github.com/".insteadOf "git@github.com:" && \
- npm install && npm run build
-WORKDIR /app
-
-# Create config directory
-RUN mkdir -p /root/.nanobot
-
-# Gateway default port
-EXPOSE 18790
-
-ENTRYPOINT ["nanobot"]
-CMD ["status"]
diff --git a/app-instance/backend-old/LICENSE b/app-instance/backend-old/LICENSE
deleted file mode 100644
index 24bdacc..0000000
--- a/app-instance/backend-old/LICENSE
+++ /dev/null
@@ -1,21 +0,0 @@
-MIT License
-
-Copyright (c) 2025 nanobot contributors
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
\ No newline at end of file
diff --git a/app-instance/backend-old/README.md b/app-instance/backend-old/README.md
deleted file mode 100644
index c161703..0000000
--- a/app-instance/backend-old/README.md
+++ /dev/null
@@ -1,470 +0,0 @@
-# Boardware Genius Backend
-
-这是 `Boardware Genius` 的后端服务仓库；当前技术命令和包名仍沿用 `nanobot`，但产品品牌按 `Boardware Genius` 表述：
-
-- `nanobot web`：单用户 FastAPI 后端，供独立前端或 `/docs` 调试使用
-- `nanobot gateway`：常驻 worker，负责渠道接入、cron、heartbeat
-- MCP 动态工具接入
-- Outlook 集成：通过外部 `BW_Outlook_Mcp` 服务接入 Microsoft Graph / Exchange EWS
-- 工作区文件、技能、插件、代理、MCP 管理等 Web API
-
-如果你后续要把它打包成 Docker 丢到服务器，这份 README 就是给开发和部署同事看的执行文档。
-
-## 这套仓库现在是什么
-
-这不是一个自带前端静态页面的全栈仓库，而是后端仓库：
-
-- Web 模式启动的是 FastAPI API 服务
-- Gateway 模式启动的是常驻 agent / channel / cron 进程
-- WhatsApp 相关逻辑依赖 `bridge/` 里的 Node 20 bridge
-- Outlook 不是仓库内置模块，而是通过外部 `BW_Outlook_Mcp` 仓库接进来
-
-更细的执行链路可以看 [workflow.md](./workflow.md)。
-
-## 目录结构
-
-```text
-.
-├── nanobot/ # Python 主体：CLI、agent、web、channels、config、MCP
-├── bridge/ # WhatsApp bridge（Node 20）
-├── tests/ # 测试
-├── Dockerfile # 当前镜像构建文件
-├── docker-compose.yml # 当前自带 compose 示例（偏 gateway / CLI）
-└── workflow.md # 运行链路说明
-```
-
-## 运行模式
-
-| 命令 | 用途 | 默认端口 | 适合谁 |
-| --- | --- | --- | --- |
-| `nanobot agent` | 本地单轮 / 交互调试 | 无 | 开发排查 |
-| `nanobot web` | 启动 FastAPI 后端 | `18080` | 独立前端、接口调试、单用户使用 |
-| `nanobot gateway` | 启动常驻 worker | 无固定 HTTP 入口 | Telegram/Slack/Email/cron/heartbeat |
-| `nanobot status` | 查看配置和 provider 状态 | 无 | 开发、运维 |
-
-注意：
-
-- 如果你是给 Web 前端提供后端，请启动 `nanobot web`，不要误用 `gateway`
-- `gateway` 当前不是对外 Web API 服务
-- `web` 和 `gateway` 都会碰到同一份 workspace / cron / MCP 状态，通常不要在同一份数据目录上无脑同时跑两套
-
-## 环境要求
-
-- Python `>=3.11`
-- 推荐使用 `uv`
-- 如果要构建 WhatsApp bridge 或使用仓库自带 Dockerfile，需要 Node.js `20`
-
-本地开发最省事的方式：
-
-```bash
-uv sync --extra dev
-```
-
-如果你不用 `uv`，也可以：
-
-```bash
-python3 -m venv .venv
-. .venv/bin/activate
-pip install -e ".[dev]"
-```
-
-## 本地快速启动
-
-### 1. 初始化配置
-
-```bash
-nanobot onboard
-```
-
-初始化后默认会生成：
-
-- 配置文件：`~/.nanobot/config.json`
-- 工作区：`~/.nanobot/workspace`
-
-### 2. 填最小配置
-
-下面是一份适合服务器环境的最小示例，重点是：
-
-- 用绝对路径的 workspace
-- 建议打开 `restrictToWorkspace`
-- 先用 API Key provider，少踩 OAuth 交互坑
-
-```json
-{
- "agents": {
- "defaults": {
- "workspace": "/root/.nanobot/workspace",
- "model": "openai/gpt-5"
- }
- },
- "providers": {
- "openai": {
- "apiKey": "sk-xxxx"
- }
- },
- "tools": {
- "restrictToWorkspace": true
- }
-}
-```
-
-如果你不是跑在容器里，把 `/root/.nanobot/workspace` 换成你自己的绝对路径。
-
-### 3. 检查配置
-
-```bash
-nanobot status
-```
-
-### 4. 本地调试 agent
-
-```bash
-nanobot agent -m "你好"
-```
-
-### 5. 启动 Web 后端
-
-```bash
-nanobot web --host 0.0.0.0 --port 18080
-```
-
-启动后可直接访问：
-
-- `http://127.0.0.1:18080/docs`
-- `http://127.0.0.1:18080/api/ping`
-
-## Web API 能力概览
-
-当前 `nanobot web` 提供的 API 大致包括：
-
-- 聊天与流式输出
-- 会话管理
-- cron 任务管理
-- skills / plugins / agents 管理
-- 工作区文件浏览、上传、下载、删除
-- MCP server 管理与测试
-- Outlook 集成状态、连接测试、连接/断开、Overview、Message Detail
-
-如果你有独立前端，这个后端就是给前端接的；如果没有前端，也可以直接走 `/docs` 调试。
-
-## Outlook MCP 集成
-
-这是当前仓库里最容易部署时踩坑的一块。
-
-### 关系先说清楚
-
-当前后端不会自己实现 Outlook 协议，它依赖外部仓库 `BW_Outlook_Mcp`：
-
-- 后端代码位置：`nanobot/web/outlook.py`
-- 默认查找逻辑：
- 1. 先看环境变量 `NANOBOT_OUTLOOK_MCP_ROOT`
- 2. 再看与本仓库同级目录的 `../BW_Outlook_Mcp`
- 3. 如果以上都没有，就尝试直接执行 PATH 里的 `bw-outlook-mcp`
-
-也就是说，部署同事必须额外把 `BW_Outlook_Mcp` 这个仓库准备好，或者把它直接安装进镜像。
-
-### 推荐的两种接法
-
-#### 方案 A：把 `BW_Outlook_Mcp` 安装进同一个 Python 环境
-
-这是生产环境更稳的方案。
-
-部署同事需要：
-
-```bash
-git clone <你们的 BW_Outlook_Mcp 仓库地址> /srv/BW_Outlook_Mcp
-cd /srv/BW_Outlook_Mcp
-pip install -e .
-```
-
-安装完成后，容器或宿主机里能直接执行：
-
-```bash
-bw-outlook-mcp --help
-```
-
-这样 Boardware Genius 就会直接用 PATH 里的 `bw-outlook-mcp`，不依赖额外挂载路径。
-
-#### 方案 B：把 `BW_Outlook_Mcp` 作为外部目录挂进来
-
-这是开发或临时部署更方便的方案。
-
-部署同事需要至少做到两件事：
-
-1. 把 `BW_Outlook_Mcp` 仓库拉到服务器
-2. 让这个目录里存在一个可执行的 `bw-outlook-mcp`
-
-最简单的约定是：
-
-```bash
-git clone <你们的 BW_Outlook_Mcp 仓库地址> /srv/BW_Outlook_Mcp
-cd /srv/BW_Outlook_Mcp
-python3 -m venv .venv
-. .venv/bin/activate
-pip install -e .
-```
-
-然后给 Boardware Genius 设置：
-
-```bash
-export NANOBOT_OUTLOOK_MCP_ROOT=/srv/BW_Outlook_Mcp
-```
-
-因为当前后端会优先寻找：
-
-```text
-$NANOBOT_OUTLOOK_MCP_ROOT/.venv/bin/bw-outlook-mcp
-```
-
-如果你挂了仓库目录但里面没有 `.venv/bin/bw-outlook-mcp`，那就必须确保 `bw-outlook-mcp` 已经在容器 PATH 里。
-
-### Outlook 的认证和配置
-
-`BW_Outlook_Mcp` 本身支持两套后端：
-
-- `graph`：Microsoft 365 / Exchange Online
-- `ews`：本地或回迁后的 Exchange Server
-
-#### Graph 登录
-
-```bash
-bw-outlook-mcp auth login-graph \
- --workspace /root/.nanobot/workspace \
- --client-id YOUR_CLIENT_ID \
- --tenant-id YOUR_TENANT_ID
-```
-
-#### EWS 配置
-
-```bash
-bw-outlook-mcp auth setup-ews \
- --workspace /root/.nanobot/workspace \
- --email you@example.com \
- --username your_username \
- --domain example.com \
- --server mail.example.com
-```
-
-如果你已经有固定 EWS URL，也可以改用：
-
-```bash
-bw-outlook-mcp auth setup-ews \
- --workspace /root/.nanobot/workspace \
- --email you@example.com \
- --username your_username \
- --service-endpoint https://mail.example.com/EWS/Exchange.asmx
-```
-
-#### 查看状态
-
-```bash
-bw-outlook-mcp auth status --workspace /root/.nanobot/workspace
-```
-
-### Outlook 状态文件会落在哪里
-
-所有 Outlook 相关状态默认都落在 workspace 下：
-
-```text
-text
to keep payload minimal. - if formatted.startswith("") and formatted.endswith("
"): - inner = formatted[3:-4] - if "<" not in inner and ">" not in inner: - return None - return formatted - - -def _build_matrix_text_content(text: str) -> dict[str, object]: - """Build Matrix m.text payload with optional HTML formatted_body.""" - content: dict[str, object] = {"msgtype": "m.text", "body": text, "m.mentions": {}} - if html := _render_markdown_html(text): - content["format"] = MATRIX_HTML_FORMAT - content["formatted_body"] = html - return content - - -class _NioLoguruHandler(logging.Handler): - """Route matrix-nio stdlib logs into Loguru.""" - - def emit(self, record: logging.LogRecord) -> None: - try: - level = logger.level(record.levelname).name - except ValueError: - level = record.levelno - frame, depth = logging.currentframe(), 2 - while frame and frame.f_code.co_filename == logging.__file__: - frame, depth = frame.f_back, depth + 1 - logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage()) - - -def _configure_nio_logging_bridge() -> None: - """Bridge matrix-nio logs to Loguru (idempotent).""" - nio_logger = logging.getLogger("nio") - if not any(isinstance(h, _NioLoguruHandler) for h in nio_logger.handlers): - nio_logger.handlers = [_NioLoguruHandler()] - nio_logger.propagate = False - - -class MatrixChannel(BaseChannel): - """Matrix (Element) channel using long-polling sync.""" - - name = "matrix" - display_name = "Matrix" - - def __init__(self, config: Any, bus: MessageBus, groq_api_key: str = ""): - super().__init__(config, bus) - self.groq_api_key = groq_api_key - self.client: AsyncClient | None = None - self._sync_task: asyncio.Task | None = None - self._typing_tasks: dict[str, asyncio.Task] = {} - self._restrict_to_workspace = False - self._workspace: Path | None = None - self._server_upload_limit_bytes: int | None = None - self._server_upload_limit_checked = False - self._sync_ready_logged = False - - async def start(self) -> None: - """Start Matrix client and begin sync loop.""" - self._running = True - _configure_nio_logging_bridge() - - store_path = get_data_dir() / "matrix-store" - store_path.mkdir(parents=True, exist_ok=True) - - self.client = AsyncClient( - homeserver=self.config.homeserver, user=self.config.user_id, - store_path=store_path, - config=AsyncClientConfig(store_sync_tokens=True, encryption_enabled=self.config.e2ee_enabled), - ) - self.client.user_id = self.config.user_id - self.client.access_token = self.config.access_token - self.client.device_id = self.config.device_id - - self._register_event_callbacks() - self._register_response_callbacks() - - if not self.config.e2ee_enabled: - logger.warning("Matrix E2EE disabled; encrypted rooms may be undecryptable.") - - if self.config.device_id: - try: - self.client.load_store() - except Exception: - logger.exception("Matrix store load failed; restart may replay recent messages.") - else: - logger.warning("Matrix device_id empty; restart may replay recent messages.") - - self._sync_task = asyncio.create_task(self._sync_loop()) - - async def stop(self) -> None: - """Stop the Matrix channel with graceful sync shutdown.""" - self._running = False - for room_id in list(self._typing_tasks): - await self._stop_typing_keepalive(room_id, clear_typing=False) - if self.client: - self.client.stop_sync_forever() - if self._sync_task: - try: - await asyncio.wait_for(asyncio.shield(self._sync_task), - timeout=self.config.sync_stop_grace_seconds) - except (asyncio.TimeoutError, asyncio.CancelledError): - self._sync_task.cancel() - try: - await self._sync_task - except asyncio.CancelledError: - pass - if self.client: - await self.client.close() - - def _is_workspace_path_allowed(self, path: Path) -> bool: - """Check path is inside workspace (when restriction enabled).""" - if not self._restrict_to_workspace or not self._workspace: - return True - try: - path.resolve(strict=False).relative_to(self._workspace) - return True - except ValueError: - return False - - def _collect_outbound_media_candidates(self, media: list[str]) -> list[Path]: - """Deduplicate and resolve outbound attachment paths.""" - seen: set[str] = set() - candidates: list[Path] = [] - for raw in media: - if not isinstance(raw, str) or not raw.strip(): - continue - path = Path(raw.strip()).expanduser() - try: - key = str(path.resolve(strict=False)) - except OSError: - key = str(path) - if key not in seen: - seen.add(key) - candidates.append(path) - return candidates - - @staticmethod - def _build_outbound_attachment_content( - *, filename: str, mime: str, size_bytes: int, - mxc_url: str, encryption_info: dict[str, Any] | None = None, - ) -> dict[str, Any]: - """Build Matrix content payload for an uploaded file/image/audio/video.""" - prefix = mime.split("/")[0] - msgtype = {"image": "m.image", "audio": "m.audio", "video": "m.video"}.get(prefix, "m.file") - content: dict[str, Any] = { - "msgtype": msgtype, "body": filename, "filename": filename, - "info": {"mimetype": mime, "size": size_bytes}, "m.mentions": {}, - } - if encryption_info: - content["file"] = {**encryption_info, "url": mxc_url} - else: - content["url"] = mxc_url - return content - - def _is_encrypted_room(self, room_id: str) -> bool: - if not self.client: - return False - room = getattr(self.client, "rooms", {}).get(room_id) - return bool(getattr(room, "encrypted", False)) - - async def _send_room_content(self, room_id: str, content: dict[str, Any]) -> None: - """Send m.room.message with E2EE options.""" - if not self.client: - return - kwargs: dict[str, Any] = {"room_id": room_id, "message_type": "m.room.message", "content": content} - if self.config.e2ee_enabled: - kwargs["ignore_unverified_devices"] = True - await self.client.room_send(**kwargs) - - async def _resolve_server_upload_limit_bytes(self) -> int | None: - """Query homeserver upload limit once per channel lifecycle.""" - if self._server_upload_limit_checked: - return self._server_upload_limit_bytes - self._server_upload_limit_checked = True - if not self.client: - return None - try: - response = await self.client.content_repository_config() - except Exception: - return None - upload_size = getattr(response, "upload_size", None) - if isinstance(upload_size, int) and upload_size > 0: - self._server_upload_limit_bytes = upload_size - return upload_size - return None - - async def _effective_media_limit_bytes(self) -> int: - """min(local config, server advertised) — 0 blocks all uploads.""" - local_limit = max(int(self.config.max_media_bytes), 0) - server_limit = await self._resolve_server_upload_limit_bytes() - if server_limit is None: - return local_limit - return min(local_limit, server_limit) if local_limit else 0 - - async def _upload_and_send_attachment( - self, room_id: str, path: Path, limit_bytes: int, - relates_to: dict[str, Any] | None = None, - ) -> str | None: - """Upload one local file to Matrix and send it as a media message. Returns failure marker or None.""" - if not self.client: - return _ATTACH_UPLOAD_FAILED.format(path.name or _DEFAULT_ATTACH_NAME) - - resolved = path.expanduser().resolve(strict=False) - filename = safe_filename(resolved.name) or _DEFAULT_ATTACH_NAME - fail = _ATTACH_UPLOAD_FAILED.format(filename) - - if not resolved.is_file() or not self._is_workspace_path_allowed(resolved): - return fail - try: - size_bytes = resolved.stat().st_size - except OSError: - return fail - if limit_bytes <= 0 or size_bytes > limit_bytes: - return _ATTACH_TOO_LARGE.format(filename) - - mime = mimetypes.guess_type(filename, strict=False)[0] or "application/octet-stream" - try: - with resolved.open("rb") as f: - upload_result = await self.client.upload( - f, content_type=mime, filename=filename, - encrypt=self.config.e2ee_enabled and self._is_encrypted_room(room_id), - filesize=size_bytes, - ) - except Exception: - return fail - - upload_response = upload_result[0] if isinstance(upload_result, tuple) else upload_result - encryption_info = upload_result[1] if isinstance(upload_result, tuple) and isinstance(upload_result[1], dict) else None - if isinstance(upload_response, UploadError): - return fail - mxc_url = getattr(upload_response, "content_uri", None) - if not isinstance(mxc_url, str) or not mxc_url.startswith("mxc://"): - return fail - - content = self._build_outbound_attachment_content( - filename=filename, mime=mime, size_bytes=size_bytes, - mxc_url=mxc_url, encryption_info=encryption_info, - ) - if relates_to: - content["m.relates_to"] = relates_to - try: - await self._send_room_content(room_id, content) - except Exception: - return fail - return None - - async def send(self, msg: OutboundMessage) -> None: - """Send outbound content; clear typing for non-progress messages.""" - if not self.client: - return - text = msg.content or "" - candidates = self._collect_outbound_media_candidates(msg.media) - relates_to = self._build_thread_relates_to(msg.metadata) - is_progress = bool((msg.metadata or {}).get("_progress")) - try: - failures: list[str] = [] - if candidates: - limit_bytes = await self._effective_media_limit_bytes() - for path in candidates: - if fail := await self._upload_and_send_attachment( - room_id=msg.chat_id, - path=path, - limit_bytes=limit_bytes, - relates_to=relates_to, - ): - failures.append(fail) - if failures: - text = f"{text.rstrip()}\n{chr(10).join(failures)}" if text.strip() else "\n".join(failures) - if text or not candidates: - content = _build_matrix_text_content(text) - if relates_to: - content["m.relates_to"] = relates_to - await self._send_room_content(msg.chat_id, content) - finally: - if not is_progress: - await self._stop_typing_keepalive(msg.chat_id, clear_typing=True) - - def _register_event_callbacks(self) -> None: - self.client.add_event_callback(self._on_message, RoomMessageText) - self.client.add_event_callback(self._on_media_message, MATRIX_MEDIA_EVENT_FILTER) - self.client.add_event_callback(self._on_room_invite, InviteEvent) - - def _register_response_callbacks(self) -> None: - self.client.add_response_callback(self._on_sync_success, SyncResponse) - self.client.add_response_callback(self._on_sync_error, SyncError) - self.client.add_response_callback(self._on_join_error, JoinError) - self.client.add_response_callback(self._on_send_error, RoomSendError) - - def _log_response_error(self, label: str, response: Any) -> None: - """Log Matrix response errors — auth errors at ERROR level, rest at WARNING.""" - code = getattr(response, "status_code", None) - is_auth = code in {"M_UNKNOWN_TOKEN", "M_FORBIDDEN", "M_UNAUTHORIZED"} - is_fatal = is_auth or getattr(response, "soft_logout", False) - (logger.error if is_fatal else logger.warning)("Matrix {} failed: {}", label, response) - - async def _on_sync_success(self, response: SyncResponse) -> None: - if self._sync_ready_logged: - return - rooms = getattr(response, "rooms", None) - joined = len(getattr(rooms, "join", {}) or {}) - invited = len(getattr(rooms, "invite", {}) or {}) - logger.info( - "Matrix sync ready: user={} device={} joined_rooms={} invited_rooms={}", - self.config.user_id, - self.config.device_id or "-", - joined, - invited, - ) - self._sync_ready_logged = True - - async def _on_sync_error(self, response: SyncError) -> None: - self._log_response_error("sync", response) - - async def _on_join_error(self, response: JoinError) -> None: - self._log_response_error("join", response) - - async def _on_send_error(self, response: RoomSendError) -> None: - self._log_response_error("send", response) - - async def _set_typing(self, room_id: str, typing: bool) -> None: - """Best-effort typing indicator update.""" - if not self.client: - return - try: - response = await self.client.room_typing(room_id=room_id, typing_state=typing, - timeout=TYPING_NOTICE_TIMEOUT_MS) - if isinstance(response, RoomTypingError): - logger.debug("Matrix typing failed for {}: {}", room_id, response) - except Exception: - pass - - async def _start_typing_keepalive(self, room_id: str) -> None: - """Start periodic typing refresh (spec-recommended keepalive).""" - await self._stop_typing_keepalive(room_id, clear_typing=False) - await self._set_typing(room_id, True) - if not self._running: - return - - async def loop() -> None: - try: - while self._running: - await asyncio.sleep(TYPING_KEEPALIVE_INTERVAL_MS / 1000) - await self._set_typing(room_id, True) - except asyncio.CancelledError: - pass - - self._typing_tasks[room_id] = asyncio.create_task(loop()) - - async def _stop_typing_keepalive(self, room_id: str, *, clear_typing: bool) -> None: - if task := self._typing_tasks.pop(room_id, None): - task.cancel() - try: - await task - except asyncio.CancelledError: - pass - if clear_typing: - await self._set_typing(room_id, False) - - async def _sync_loop(self) -> None: - while self._running: - try: - await self.client.sync_forever(timeout=30000, full_state=True) - except asyncio.CancelledError: - break - except Exception: - await asyncio.sleep(2) - - async def _on_room_invite(self, room: MatrixRoom, event: InviteEvent) -> None: - if self.is_allowed(event.sender): - await self.client.join(room.room_id) - - def _is_direct_room(self, room: MatrixRoom) -> bool: - count = getattr(room, "member_count", None) - return isinstance(count, int) and count <= 2 - - def _is_bot_mentioned(self, event: RoomMessage) -> bool: - """Check m.mentions payload for bot mention.""" - source = getattr(event, "source", None) - if not isinstance(source, dict): - return False - mentions = (source.get("content") or {}).get("m.mentions") - if not isinstance(mentions, dict): - return False - user_ids = mentions.get("user_ids") - if isinstance(user_ids, list) and self.config.user_id in user_ids: - return True - return bool(self.config.allow_room_mentions and mentions.get("room") is True) - - def _should_process_message(self, room: MatrixRoom, event: RoomMessage) -> bool: - """Apply sender and room policy checks.""" - if not self.is_allowed(event.sender): - return False - if self._is_direct_room(room): - return True - policy = self.config.group_policy - if policy == "open": - return True - if policy == "allowlist": - return room.room_id in (self.config.group_allow_from or []) - if policy == "mention": - return self._is_bot_mentioned(event) - return False - - def _media_dir(self) -> Path: - return get_media_dir("matrix") - - async def transcribe_audio(self, file_path: str) -> str: - """Best-effort audio transcription for inbound Matrix voice/audio messages.""" - try: - return await GroqTranscriptionProvider(api_key=self.groq_api_key).transcribe(file_path) - except Exception: - logger.exception("Matrix audio transcription failed") - return "" - - @staticmethod - def _event_source_content(event: RoomMessage) -> dict[str, Any]: - source = getattr(event, "source", None) - if not isinstance(source, dict): - return {} - content = source.get("content") - return content if isinstance(content, dict) else {} - - def _event_thread_root_id(self, event: RoomMessage) -> str | None: - relates_to = self._event_source_content(event).get("m.relates_to") - if not isinstance(relates_to, dict) or relates_to.get("rel_type") != "m.thread": - return None - root_id = relates_to.get("event_id") - return root_id if isinstance(root_id, str) and root_id else None - - def _thread_metadata(self, event: RoomMessage) -> dict[str, str] | None: - if not (root_id := self._event_thread_root_id(event)): - return None - meta: dict[str, str] = {"thread_root_event_id": root_id} - if isinstance(reply_to := getattr(event, "event_id", None), str) and reply_to: - meta["thread_reply_to_event_id"] = reply_to - return meta - - @staticmethod - def _build_thread_relates_to(metadata: dict[str, Any] | None) -> dict[str, Any] | None: - if not metadata: - return None - root_id = metadata.get("thread_root_event_id") - if not isinstance(root_id, str) or not root_id: - return None - reply_to = metadata.get("thread_reply_to_event_id") or metadata.get("event_id") - if not isinstance(reply_to, str) or not reply_to: - return None - return {"rel_type": "m.thread", "event_id": root_id, - "m.in_reply_to": {"event_id": reply_to}, "is_falling_back": True} - - def _event_attachment_type(self, event: MatrixMediaEvent) -> str: - msgtype = self._event_source_content(event).get("msgtype") - return _MSGTYPE_MAP.get(msgtype, "file") - - @staticmethod - def _is_encrypted_media_event(event: MatrixMediaEvent) -> bool: - return (isinstance(getattr(event, "key", None), dict) - and isinstance(getattr(event, "hashes", None), dict) - and isinstance(getattr(event, "iv", None), str)) - - def _event_declared_size_bytes(self, event: MatrixMediaEvent) -> int | None: - info = self._event_source_content(event).get("info") - size = info.get("size") if isinstance(info, dict) else None - return size if isinstance(size, int) and size >= 0 else None - - def _event_mime(self, event: MatrixMediaEvent) -> str | None: - info = self._event_source_content(event).get("info") - if isinstance(info, dict) and isinstance(m := info.get("mimetype"), str) and m: - return m - m = getattr(event, "mimetype", None) - return m if isinstance(m, str) and m else None - - def _event_filename(self, event: MatrixMediaEvent, attachment_type: str) -> str: - body = getattr(event, "body", None) - if isinstance(body, str) and body.strip(): - if candidate := safe_filename(Path(body).name): - return candidate - return _DEFAULT_ATTACH_NAME if attachment_type == "file" else attachment_type - - def _build_attachment_path(self, event: MatrixMediaEvent, attachment_type: str, - filename: str, mime: str | None) -> Path: - safe_name = safe_filename(Path(filename).name) or _DEFAULT_ATTACH_NAME - suffix = Path(safe_name).suffix - if not suffix and mime: - if guessed := mimetypes.guess_extension(mime, strict=False): - safe_name, suffix = f"{safe_name}{guessed}", guessed - stem = (Path(safe_name).stem or attachment_type)[:72] - suffix = suffix[:16] - event_id = safe_filename(str(getattr(event, "event_id", "") or "evt").lstrip("$")) - event_prefix = (event_id[:24] or "evt").strip("_") - return self._media_dir() / f"{event_prefix}_{stem}{suffix}" - - async def _download_media_bytes(self, mxc_url: str) -> bytes | None: - if not self.client: - return None - response = await self.client.download(mxc=mxc_url) - if isinstance(response, DownloadError): - logger.warning("Matrix download failed for {}: {}", mxc_url, response) - return None - body = getattr(response, "body", None) - if isinstance(body, (bytes, bytearray)): - return bytes(body) - if isinstance(response, MemoryDownloadResponse): - return bytes(response.body) - if isinstance(body, (str, Path)): - path = Path(body) - if path.is_file(): - try: - return path.read_bytes() - except OSError: - return None - return None - - def _decrypt_media_bytes(self, event: MatrixMediaEvent, ciphertext: bytes) -> bytes | None: - key_obj, hashes, iv = getattr(event, "key", None), getattr(event, "hashes", None), getattr(event, "iv", None) - key = key_obj.get("k") if isinstance(key_obj, dict) else None - sha256 = hashes.get("sha256") if isinstance(hashes, dict) else None - if not all(isinstance(v, str) for v in (key, sha256, iv)): - return None - try: - return decrypt_attachment(ciphertext, key, sha256, iv) - except (EncryptionError, ValueError, TypeError): - logger.warning("Matrix decrypt failed for event {}", getattr(event, "event_id", "")) - return None - - async def _fetch_media_attachment( - self, room: MatrixRoom, event: MatrixMediaEvent, - ) -> tuple[dict[str, Any] | None, str]: - """Download, decrypt if needed, and persist a Matrix attachment.""" - atype = self._event_attachment_type(event) - mime = self._event_mime(event) - filename = self._event_filename(event, atype) - mxc_url = getattr(event, "url", None) - fail = _ATTACH_FAILED.format(filename) - - if not isinstance(mxc_url, str) or not mxc_url.startswith("mxc://"): - return None, fail - - limit_bytes = await self._effective_media_limit_bytes() - declared = self._event_declared_size_bytes(event) - if declared is not None and declared > limit_bytes: - return None, _ATTACH_TOO_LARGE.format(filename) - - downloaded = await self._download_media_bytes(mxc_url) - if downloaded is None: - return None, fail - - encrypted = self._is_encrypted_media_event(event) - data = downloaded - if encrypted: - if (data := self._decrypt_media_bytes(event, downloaded)) is None: - return None, fail - - if len(data) > limit_bytes: - return None, _ATTACH_TOO_LARGE.format(filename) - - path = self._build_attachment_path(event, atype, filename, mime) - try: - path.write_bytes(data) - except OSError: - return None, fail - - attachment = { - "type": atype, "mime": mime, "filename": filename, - "event_id": str(getattr(event, "event_id", "") or ""), - "encrypted": encrypted, "size_bytes": len(data), - "path": str(path), "mxc_url": mxc_url, - } - return attachment, _ATTACH_MARKER.format(path) - - def _base_metadata(self, room: MatrixRoom, event: RoomMessage) -> dict[str, Any]: - """Build common metadata for text and media handlers.""" - meta: dict[str, Any] = {"room": getattr(room, "display_name", room.room_id)} - if isinstance(eid := getattr(event, "event_id", None), str) and eid: - meta["event_id"] = eid - if thread := self._thread_metadata(event): - meta.update(thread) - return meta - - async def _on_message(self, room: MatrixRoom, event: RoomMessageText) -> None: - if event.sender == self.config.user_id or not self._should_process_message(room, event): - return - await self._start_typing_keepalive(room.room_id) - try: - await self._handle_message( - sender_id=event.sender, chat_id=room.room_id, - content=event.body, metadata=self._base_metadata(room, event), - ) - except Exception: - await self._stop_typing_keepalive(room.room_id, clear_typing=True) - raise - - async def _on_media_message(self, room: MatrixRoom, event: MatrixMediaEvent) -> None: - if event.sender == self.config.user_id or not self._should_process_message(room, event): - return - attachment, marker = await self._fetch_media_attachment(room, event) - parts: list[str] = [] - if isinstance(body := getattr(event, "body", None), str) and body.strip(): - parts.append(body.strip()) - - if attachment and attachment.get("type") == "audio": - transcription = await self.transcribe_audio(attachment["path"]) - if transcription: - parts.append(f"[transcription: {transcription}]") - else: - parts.append(marker) - elif marker: - parts.append(marker) - - await self._start_typing_keepalive(room.room_id) - try: - meta = self._base_metadata(room, event) - meta["attachments"] = [] - if attachment: - meta["attachments"] = [attachment] - await self._handle_message( - sender_id=event.sender, chat_id=room.room_id, - content="\n".join(parts), - media=[attachment["path"]] if attachment else [], - metadata=meta, - ) - except Exception: - await self._stop_typing_keepalive(room.room_id, clear_typing=True) - raise diff --git a/app-instance/backend-old/nanobot/channels/mochat.py b/app-instance/backend-old/nanobot/channels/mochat.py deleted file mode 100644 index e762dfd..0000000 --- a/app-instance/backend-old/nanobot/channels/mochat.py +++ /dev/null @@ -1,895 +0,0 @@ -"""Mochat channel implementation using Socket.IO with HTTP polling fallback.""" - -from __future__ import annotations - -import asyncio -import json -from collections import deque -from dataclasses import dataclass, field -from datetime import datetime -from typing import Any - -import httpx -from loguru import logger - -from nanobot.bus.events import OutboundMessage -from nanobot.bus.queue import MessageBus -from nanobot.channels.base import BaseChannel -from nanobot.config.schema import MochatConfig -from nanobot.utils.helpers import get_data_path - -try: - import socketio - SOCKETIO_AVAILABLE = True -except ImportError: - socketio = None - SOCKETIO_AVAILABLE = False - -try: - import msgpack # noqa: F401 - MSGPACK_AVAILABLE = True -except ImportError: - MSGPACK_AVAILABLE = False - -MAX_SEEN_MESSAGE_IDS = 2000 -CURSOR_SAVE_DEBOUNCE_S = 0.5 - - -# --------------------------------------------------------------------------- -# Data classes -# --------------------------------------------------------------------------- - -@dataclass -class MochatBufferedEntry: - """Buffered inbound entry for delayed dispatch.""" - raw_body: str - author: str - sender_name: str = "" - sender_username: str = "" - timestamp: int | None = None - message_id: str = "" - group_id: str = "" - - -@dataclass -class DelayState: - """Per-target delayed message state.""" - entries: list[MochatBufferedEntry] = field(default_factory=list) - lock: asyncio.Lock = field(default_factory=asyncio.Lock) - timer: asyncio.Task | None = None - - -@dataclass -class MochatTarget: - """Outbound target resolution result.""" - id: str - is_panel: bool - - -# --------------------------------------------------------------------------- -# Pure helpers -# --------------------------------------------------------------------------- - -def _safe_dict(value: Any) -> dict: - """Return *value* if it's a dict, else empty dict.""" - return value if isinstance(value, dict) else {} - - -def _str_field(src: dict, *keys: str) -> str: - """Return the first non-empty str value found for *keys*, stripped.""" - for k in keys: - v = src.get(k) - if isinstance(v, str) and v.strip(): - return v.strip() - return "" - - -def _make_synthetic_event( - message_id: str, author: str, content: Any, - meta: Any, group_id: str, converse_id: str, - timestamp: Any = None, *, author_info: Any = None, -) -> dict[str, Any]: - """Build a synthetic ``message.add`` event dict.""" - payload: dict[str, Any] = { - "messageId": message_id, "author": author, - "content": content, "meta": _safe_dict(meta), - "groupId": group_id, "converseId": converse_id, - } - if author_info is not None: - payload["authorInfo"] = _safe_dict(author_info) - return { - "type": "message.add", - "timestamp": timestamp or datetime.utcnow().isoformat(), - "payload": payload, - } - - -def normalize_mochat_content(content: Any) -> str: - """Normalize content payload to text.""" - if isinstance(content, str): - return content.strip() - if content is None: - return "" - try: - return json.dumps(content, ensure_ascii=False) - except TypeError: - return str(content) - - -def resolve_mochat_target(raw: str) -> MochatTarget: - """Resolve id and target kind from user-provided target string.""" - trimmed = (raw or "").strip() - if not trimmed: - return MochatTarget(id="", is_panel=False) - - lowered = trimmed.lower() - cleaned, forced_panel = trimmed, False - for prefix in ("mochat:", "group:", "channel:", "panel:"): - if lowered.startswith(prefix): - cleaned = trimmed[len(prefix):].strip() - forced_panel = prefix in {"group:", "channel:", "panel:"} - break - - if not cleaned: - return MochatTarget(id="", is_panel=False) - return MochatTarget(id=cleaned, is_panel=forced_panel or not cleaned.startswith("session_")) - - -def extract_mention_ids(value: Any) -> list[str]: - """Extract mention ids from heterogeneous mention payload.""" - if not isinstance(value, list): - return [] - ids: list[str] = [] - for item in value: - if isinstance(item, str): - if item.strip(): - ids.append(item.strip()) - elif isinstance(item, dict): - for key in ("id", "userId", "_id"): - candidate = item.get(key) - if isinstance(candidate, str) and candidate.strip(): - ids.append(candidate.strip()) - break - return ids - - -def resolve_was_mentioned(payload: dict[str, Any], agent_user_id: str) -> bool: - """Resolve mention state from payload metadata and text fallback.""" - meta = payload.get("meta") - if isinstance(meta, dict): - if meta.get("mentioned") is True or meta.get("wasMentioned") is True: - return True - for f in ("mentions", "mentionIds", "mentionedUserIds", "mentionedUsers"): - if agent_user_id and agent_user_id in extract_mention_ids(meta.get(f)): - return True - if not agent_user_id: - return False - content = payload.get("content") - if not isinstance(content, str) or not content: - return False - return f"<@{agent_user_id}>" in content or f"@{agent_user_id}" in content - - -def resolve_require_mention(config: MochatConfig, session_id: str, group_id: str) -> bool: - """Resolve mention requirement for group/panel conversations.""" - groups = config.groups or {} - for key in (group_id, session_id, "*"): - if key and key in groups: - return bool(groups[key].require_mention) - return bool(config.mention.require_in_groups) - - -def build_buffered_body(entries: list[MochatBufferedEntry], is_group: bool) -> str: - """Build text body from one or more buffered entries.""" - if not entries: - return "" - if len(entries) == 1: - return entries[0].raw_body - lines: list[str] = [] - for entry in entries: - if not entry.raw_body: - continue - if is_group: - label = entry.sender_name.strip() or entry.sender_username.strip() or entry.author - if label: - lines.append(f"{label}: {entry.raw_body}") - continue - lines.append(entry.raw_body) - return "\n".join(lines).strip() - - -def parse_timestamp(value: Any) -> int | None: - """Parse event timestamp to epoch milliseconds.""" - if not isinstance(value, str) or not value.strip(): - return None - try: - return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000) - except ValueError: - return None - - -# --------------------------------------------------------------------------- -# Channel -# --------------------------------------------------------------------------- - -class MochatChannel(BaseChannel): - """Mochat channel using socket.io with fallback polling workers.""" - - name = "mochat" - - def __init__(self, config: MochatConfig, bus: MessageBus): - super().__init__(config, bus) - self.config: MochatConfig = config - self._http: httpx.AsyncClient | None = None - self._socket: Any = None - self._ws_connected = self._ws_ready = False - - self._state_dir = get_data_path() / "mochat" - self._cursor_path = self._state_dir / "session_cursors.json" - self._session_cursor: dict[str, int] = {} - self._cursor_save_task: asyncio.Task | None = None - - self._session_set: set[str] = set() - self._panel_set: set[str] = set() - self._auto_discover_sessions = self._auto_discover_panels = False - - self._cold_sessions: set[str] = set() - self._session_by_converse: dict[str, str] = {} - - self._seen_set: dict[str, set[str]] = {} - self._seen_queue: dict[str, deque[str]] = {} - self._delay_states: dict[str, DelayState] = {} - - self._fallback_mode = False - self._session_fallback_tasks: dict[str, asyncio.Task] = {} - self._panel_fallback_tasks: dict[str, asyncio.Task] = {} - self._refresh_task: asyncio.Task | None = None - self._target_locks: dict[str, asyncio.Lock] = {} - - # ---- lifecycle --------------------------------------------------------- - - async def start(self) -> None: - """Start Mochat channel workers and websocket connection.""" - if not self.config.claw_token: - logger.error("Mochat claw_token not configured") - return - - self._running = True - self._http = httpx.AsyncClient(timeout=30.0) - self._state_dir.mkdir(parents=True, exist_ok=True) - await self._load_session_cursors() - self._seed_targets_from_config() - await self._refresh_targets(subscribe_new=False) - - if not await self._start_socket_client(): - await self._ensure_fallback_workers() - - self._refresh_task = asyncio.create_task(self._refresh_loop()) - while self._running: - await asyncio.sleep(1) - - async def stop(self) -> None: - """Stop all workers and clean up resources.""" - self._running = False - if self._refresh_task: - self._refresh_task.cancel() - self._refresh_task = None - - await self._stop_fallback_workers() - await self._cancel_delay_timers() - - if self._socket: - try: - await self._socket.disconnect() - except Exception: - pass - self._socket = None - - if self._cursor_save_task: - self._cursor_save_task.cancel() - self._cursor_save_task = None - await self._save_session_cursors() - - if self._http: - await self._http.aclose() - self._http = None - self._ws_connected = self._ws_ready = False - - async def send(self, msg: OutboundMessage) -> None: - """Send outbound message to session or panel.""" - if not self.config.claw_token: - logger.warning("Mochat claw_token missing, skip send") - return - - parts = ([msg.content.strip()] if msg.content and msg.content.strip() else []) - if msg.media: - parts.extend(m for m in msg.media if isinstance(m, str) and m.strip()) - content = "\n".join(parts).strip() - if not content: - return - - target = resolve_mochat_target(msg.chat_id) - if not target.id: - logger.warning("Mochat outbound target is empty") - return - - is_panel = (target.is_panel or target.id in self._panel_set) and not target.id.startswith("session_") - try: - if is_panel: - await self._api_send("/api/claw/groups/panels/send", "panelId", target.id, - content, msg.reply_to, self._read_group_id(msg.metadata)) - else: - await self._api_send("/api/claw/sessions/send", "sessionId", target.id, - content, msg.reply_to) - except Exception as e: - logger.error("Failed to send Mochat message: {}", e) - - # ---- config / init helpers --------------------------------------------- - - def _seed_targets_from_config(self) -> None: - sessions, self._auto_discover_sessions = self._normalize_id_list(self.config.sessions) - panels, self._auto_discover_panels = self._normalize_id_list(self.config.panels) - self._session_set.update(sessions) - self._panel_set.update(panels) - for sid in sessions: - if sid not in self._session_cursor: - self._cold_sessions.add(sid) - - @staticmethod - def _normalize_id_list(values: list[str]) -> tuple[list[str], bool]: - cleaned = [str(v).strip() for v in values if str(v).strip()] - return sorted({v for v in cleaned if v != "*"}), "*" in cleaned - - # ---- websocket --------------------------------------------------------- - - async def _start_socket_client(self) -> bool: - if not SOCKETIO_AVAILABLE: - logger.warning("python-socketio not installed, Mochat using polling fallback") - return False - - serializer = "default" - if not self.config.socket_disable_msgpack: - if MSGPACK_AVAILABLE: - serializer = "msgpack" - else: - logger.warning("msgpack not installed but socket_disable_msgpack=false; using JSON") - - client = socketio.AsyncClient( - reconnection=True, - reconnection_attempts=self.config.max_retry_attempts or None, - reconnection_delay=max(0.1, self.config.socket_reconnect_delay_ms / 1000.0), - reconnection_delay_max=max(0.1, self.config.socket_max_reconnect_delay_ms / 1000.0), - logger=False, engineio_logger=False, serializer=serializer, - ) - - @client.event - async def connect() -> None: - self._ws_connected, self._ws_ready = True, False - logger.info("Mochat websocket connected") - subscribed = await self._subscribe_all() - self._ws_ready = subscribed - await (self._stop_fallback_workers() if subscribed else self._ensure_fallback_workers()) - - @client.event - async def disconnect() -> None: - if not self._running: - return - self._ws_connected = self._ws_ready = False - logger.warning("Mochat websocket disconnected") - await self._ensure_fallback_workers() - - @client.event - async def connect_error(data: Any) -> None: - logger.error("Mochat websocket connect error: {}", data) - - @client.on("claw.session.events") - async def on_session_events(payload: dict[str, Any]) -> None: - await self._handle_watch_payload(payload, "session") - - @client.on("claw.panel.events") - async def on_panel_events(payload: dict[str, Any]) -> None: - await self._handle_watch_payload(payload, "panel") - - for ev in ("notify:chat.inbox.append", "notify:chat.message.add", - "notify:chat.message.update", "notify:chat.message.recall", - "notify:chat.message.delete"): - client.on(ev, self._build_notify_handler(ev)) - - socket_url = (self.config.socket_url or self.config.base_url).strip().rstrip("/") - socket_path = (self.config.socket_path or "/socket.io").strip().lstrip("/") - - try: - self._socket = client - await client.connect( - socket_url, transports=["websocket"], socketio_path=socket_path, - auth={"token": self.config.claw_token}, - wait_timeout=max(1.0, self.config.socket_connect_timeout_ms / 1000.0), - ) - return True - except Exception as e: - logger.error("Failed to connect Mochat websocket: {}", e) - try: - await client.disconnect() - except Exception: - pass - self._socket = None - return False - - def _build_notify_handler(self, event_name: str): - async def handler(payload: Any) -> None: - if event_name == "notify:chat.inbox.append": - await self._handle_notify_inbox_append(payload) - elif event_name.startswith("notify:chat.message."): - await self._handle_notify_chat_message(payload) - return handler - - # ---- subscribe --------------------------------------------------------- - - async def _subscribe_all(self) -> bool: - ok = await self._subscribe_sessions(sorted(self._session_set)) - ok = await self._subscribe_panels(sorted(self._panel_set)) and ok - if self._auto_discover_sessions or self._auto_discover_panels: - await self._refresh_targets(subscribe_new=True) - return ok - - async def _subscribe_sessions(self, session_ids: list[str]) -> bool: - if not session_ids: - return True - for sid in session_ids: - if sid not in self._session_cursor: - self._cold_sessions.add(sid) - - ack = await self._socket_call("com.claw.im.subscribeSessions", { - "sessionIds": session_ids, "cursors": self._session_cursor, - "limit": self.config.watch_limit, - }) - if not ack.get("result"): - logger.error("Mochat subscribeSessions failed: {}", ack.get('message', 'unknown error')) - return False - - data = ack.get("data") - items: list[dict[str, Any]] = [] - if isinstance(data, list): - items = [i for i in data if isinstance(i, dict)] - elif isinstance(data, dict): - sessions = data.get("sessions") - if isinstance(sessions, list): - items = [i for i in sessions if isinstance(i, dict)] - elif "sessionId" in data: - items = [data] - for p in items: - await self._handle_watch_payload(p, "session") - return True - - async def _subscribe_panels(self, panel_ids: list[str]) -> bool: - if not self._auto_discover_panels and not panel_ids: - return True - ack = await self._socket_call("com.claw.im.subscribePanels", {"panelIds": panel_ids}) - if not ack.get("result"): - logger.error("Mochat subscribePanels failed: {}", ack.get('message', 'unknown error')) - return False - return True - - async def _socket_call(self, event_name: str, payload: dict[str, Any]) -> dict[str, Any]: - if not self._socket: - return {"result": False, "message": "socket not connected"} - try: - raw = await self._socket.call(event_name, payload, timeout=10) - except Exception as e: - return {"result": False, "message": str(e)} - return raw if isinstance(raw, dict) else {"result": True, "data": raw} - - # ---- refresh / discovery ----------------------------------------------- - - async def _refresh_loop(self) -> None: - interval_s = max(1.0, self.config.refresh_interval_ms / 1000.0) - while self._running: - await asyncio.sleep(interval_s) - try: - await self._refresh_targets(subscribe_new=self._ws_ready) - except Exception as e: - logger.warning("Mochat refresh failed: {}", e) - if self._fallback_mode: - await self._ensure_fallback_workers() - - async def _refresh_targets(self, subscribe_new: bool) -> None: - if self._auto_discover_sessions: - await self._refresh_sessions_directory(subscribe_new) - if self._auto_discover_panels: - await self._refresh_panels(subscribe_new) - - async def _refresh_sessions_directory(self, subscribe_new: bool) -> None: - try: - response = await self._post_json("/api/claw/sessions/list", {}) - except Exception as e: - logger.warning("Mochat listSessions failed: {}", e) - return - - sessions = response.get("sessions") - if not isinstance(sessions, list): - return - - new_ids: list[str] = [] - for s in sessions: - if not isinstance(s, dict): - continue - sid = _str_field(s, "sessionId") - if not sid: - continue - if sid not in self._session_set: - self._session_set.add(sid) - new_ids.append(sid) - if sid not in self._session_cursor: - self._cold_sessions.add(sid) - cid = _str_field(s, "converseId") - if cid: - self._session_by_converse[cid] = sid - - if not new_ids: - return - if self._ws_ready and subscribe_new: - await self._subscribe_sessions(new_ids) - if self._fallback_mode: - await self._ensure_fallback_workers() - - async def _refresh_panels(self, subscribe_new: bool) -> None: - try: - response = await self._post_json("/api/claw/groups/get", {}) - except Exception as e: - logger.warning("Mochat getWorkspaceGroup failed: {}", e) - return - - raw_panels = response.get("panels") - if not isinstance(raw_panels, list): - return - - new_ids: list[str] = [] - for p in raw_panels: - if not isinstance(p, dict): - continue - pt = p.get("type") - if isinstance(pt, int) and pt != 0: - continue - pid = _str_field(p, "id", "_id") - if pid and pid not in self._panel_set: - self._panel_set.add(pid) - new_ids.append(pid) - - if not new_ids: - return - if self._ws_ready and subscribe_new: - await self._subscribe_panels(new_ids) - if self._fallback_mode: - await self._ensure_fallback_workers() - - # ---- fallback workers -------------------------------------------------- - - async def _ensure_fallback_workers(self) -> None: - if not self._running: - return - self._fallback_mode = True - for sid in sorted(self._session_set): - t = self._session_fallback_tasks.get(sid) - if not t or t.done(): - self._session_fallback_tasks[sid] = asyncio.create_task(self._session_watch_worker(sid)) - for pid in sorted(self._panel_set): - t = self._panel_fallback_tasks.get(pid) - if not t or t.done(): - self._panel_fallback_tasks[pid] = asyncio.create_task(self._panel_poll_worker(pid)) - - async def _stop_fallback_workers(self) -> None: - self._fallback_mode = False - tasks = [*self._session_fallback_tasks.values(), *self._panel_fallback_tasks.values()] - for t in tasks: - t.cancel() - if tasks: - await asyncio.gather(*tasks, return_exceptions=True) - self._session_fallback_tasks.clear() - self._panel_fallback_tasks.clear() - - async def _session_watch_worker(self, session_id: str) -> None: - while self._running and self._fallback_mode: - try: - payload = await self._post_json("/api/claw/sessions/watch", { - "sessionId": session_id, "cursor": self._session_cursor.get(session_id, 0), - "timeoutMs": self.config.watch_timeout_ms, "limit": self.config.watch_limit, - }) - await self._handle_watch_payload(payload, "session") - except asyncio.CancelledError: - break - except Exception as e: - logger.warning("Mochat watch fallback error ({}): {}", session_id, e) - await asyncio.sleep(max(0.1, self.config.retry_delay_ms / 1000.0)) - - async def _panel_poll_worker(self, panel_id: str) -> None: - sleep_s = max(1.0, self.config.refresh_interval_ms / 1000.0) - while self._running and self._fallback_mode: - try: - resp = await self._post_json("/api/claw/groups/panels/messages", { - "panelId": panel_id, "limit": min(100, max(1, self.config.watch_limit)), - }) - msgs = resp.get("messages") - if isinstance(msgs, list): - for m in reversed(msgs): - if not isinstance(m, dict): - continue - evt = _make_synthetic_event( - message_id=str(m.get("messageId") or ""), - author=str(m.get("author") or ""), - content=m.get("content"), - meta=m.get("meta"), group_id=str(resp.get("groupId") or ""), - converse_id=panel_id, timestamp=m.get("createdAt"), - author_info=m.get("authorInfo"), - ) - await self._process_inbound_event(panel_id, evt, "panel") - except asyncio.CancelledError: - break - except Exception as e: - logger.warning("Mochat panel polling error ({}): {}", panel_id, e) - await asyncio.sleep(sleep_s) - - # ---- inbound event processing ------------------------------------------ - - async def _handle_watch_payload(self, payload: dict[str, Any], target_kind: str) -> None: - if not isinstance(payload, dict): - return - target_id = _str_field(payload, "sessionId") - if not target_id: - return - - lock = self._target_locks.setdefault(f"{target_kind}:{target_id}", asyncio.Lock()) - async with lock: - prev = self._session_cursor.get(target_id, 0) if target_kind == "session" else 0 - pc = payload.get("cursor") - if target_kind == "session" and isinstance(pc, int) and pc >= 0: - self._mark_session_cursor(target_id, pc) - - raw_events = payload.get("events") - if not isinstance(raw_events, list): - return - if target_kind == "session" and target_id in self._cold_sessions: - self._cold_sessions.discard(target_id) - return - - for event in raw_events: - if not isinstance(event, dict): - continue - seq = event.get("seq") - if target_kind == "session" and isinstance(seq, int) and seq > self._session_cursor.get(target_id, prev): - self._mark_session_cursor(target_id, seq) - if event.get("type") == "message.add": - await self._process_inbound_event(target_id, event, target_kind) - - async def _process_inbound_event(self, target_id: str, event: dict[str, Any], target_kind: str) -> None: - payload = event.get("payload") - if not isinstance(payload, dict): - return - - author = _str_field(payload, "author") - if not author or (self.config.agent_user_id and author == self.config.agent_user_id): - return - if not self.is_allowed(author): - return - - message_id = _str_field(payload, "messageId") - seen_key = f"{target_kind}:{target_id}" - if message_id and self._remember_message_id(seen_key, message_id): - return - - raw_body = normalize_mochat_content(payload.get("content")) or "[empty message]" - ai = _safe_dict(payload.get("authorInfo")) - sender_name = _str_field(ai, "nickname", "email") - sender_username = _str_field(ai, "agentId") - - group_id = _str_field(payload, "groupId") - is_group = bool(group_id) - was_mentioned = resolve_was_mentioned(payload, self.config.agent_user_id) - require_mention = target_kind == "panel" and is_group and resolve_require_mention(self.config, target_id, group_id) - use_delay = target_kind == "panel" and self.config.reply_delay_mode == "non-mention" - - if require_mention and not was_mentioned and not use_delay: - return - - entry = MochatBufferedEntry( - raw_body=raw_body, author=author, sender_name=sender_name, - sender_username=sender_username, timestamp=parse_timestamp(event.get("timestamp")), - message_id=message_id, group_id=group_id, - ) - - if use_delay: - delay_key = seen_key - if was_mentioned: - await self._flush_delayed_entries(delay_key, target_id, target_kind, "mention", entry) - else: - await self._enqueue_delayed_entry(delay_key, target_id, target_kind, entry) - return - - await self._dispatch_entries(target_id, target_kind, [entry], was_mentioned) - - # ---- dedup / buffering ------------------------------------------------- - - def _remember_message_id(self, key: str, message_id: str) -> bool: - seen_set = self._seen_set.setdefault(key, set()) - seen_queue = self._seen_queue.setdefault(key, deque()) - if message_id in seen_set: - return True - seen_set.add(message_id) - seen_queue.append(message_id) - while len(seen_queue) > MAX_SEEN_MESSAGE_IDS: - seen_set.discard(seen_queue.popleft()) - return False - - async def _enqueue_delayed_entry(self, key: str, target_id: str, target_kind: str, entry: MochatBufferedEntry) -> None: - state = self._delay_states.setdefault(key, DelayState()) - async with state.lock: - state.entries.append(entry) - if state.timer: - state.timer.cancel() - state.timer = asyncio.create_task(self._delay_flush_after(key, target_id, target_kind)) - - async def _delay_flush_after(self, key: str, target_id: str, target_kind: str) -> None: - await asyncio.sleep(max(0, self.config.reply_delay_ms) / 1000.0) - await self._flush_delayed_entries(key, target_id, target_kind, "timer", None) - - async def _flush_delayed_entries(self, key: str, target_id: str, target_kind: str, reason: str, entry: MochatBufferedEntry | None) -> None: - state = self._delay_states.setdefault(key, DelayState()) - async with state.lock: - if entry: - state.entries.append(entry) - current = asyncio.current_task() - if state.timer and state.timer is not current: - state.timer.cancel() - state.timer = None - entries = state.entries[:] - state.entries.clear() - if entries: - await self._dispatch_entries(target_id, target_kind, entries, reason == "mention") - - async def _dispatch_entries(self, target_id: str, target_kind: str, entries: list[MochatBufferedEntry], was_mentioned: bool) -> None: - if not entries: - return - last = entries[-1] - is_group = bool(last.group_id) - body = build_buffered_body(entries, is_group) or "[empty message]" - await self._handle_message( - sender_id=last.author, chat_id=target_id, content=body, - metadata={ - "message_id": last.message_id, "timestamp": last.timestamp, - "is_group": is_group, "group_id": last.group_id, - "sender_name": last.sender_name, "sender_username": last.sender_username, - "target_kind": target_kind, "was_mentioned": was_mentioned, - "buffered_count": len(entries), - }, - ) - - async def _cancel_delay_timers(self) -> None: - for state in self._delay_states.values(): - if state.timer: - state.timer.cancel() - self._delay_states.clear() - - # ---- notify handlers --------------------------------------------------- - - async def _handle_notify_chat_message(self, payload: Any) -> None: - if not isinstance(payload, dict): - return - group_id = _str_field(payload, "groupId") - panel_id = _str_field(payload, "converseId", "panelId") - if not group_id or not panel_id: - return - if self._panel_set and panel_id not in self._panel_set: - return - - evt = _make_synthetic_event( - message_id=str(payload.get("_id") or payload.get("messageId") or ""), - author=str(payload.get("author") or ""), - content=payload.get("content"), meta=payload.get("meta"), - group_id=group_id, converse_id=panel_id, - timestamp=payload.get("createdAt"), author_info=payload.get("authorInfo"), - ) - await self._process_inbound_event(panel_id, evt, "panel") - - async def _handle_notify_inbox_append(self, payload: Any) -> None: - if not isinstance(payload, dict) or payload.get("type") != "message": - return - detail = payload.get("payload") - if not isinstance(detail, dict): - return - if _str_field(detail, "groupId"): - return - converse_id = _str_field(detail, "converseId") - if not converse_id: - return - - session_id = self._session_by_converse.get(converse_id) - if not session_id: - await self._refresh_sessions_directory(self._ws_ready) - session_id = self._session_by_converse.get(converse_id) - if not session_id: - return - - evt = _make_synthetic_event( - message_id=str(detail.get("messageId") or payload.get("_id") or ""), - author=str(detail.get("messageAuthor") or ""), - content=str(detail.get("messagePlainContent") or detail.get("messageSnippet") or ""), - meta={"source": "notify:chat.inbox.append", "converseId": converse_id}, - group_id="", converse_id=converse_id, timestamp=payload.get("createdAt"), - ) - await self._process_inbound_event(session_id, evt, "session") - - # ---- cursor persistence ------------------------------------------------ - - def _mark_session_cursor(self, session_id: str, cursor: int) -> None: - if cursor < 0 or cursor < self._session_cursor.get(session_id, 0): - return - self._session_cursor[session_id] = cursor - if not self._cursor_save_task or self._cursor_save_task.done(): - self._cursor_save_task = asyncio.create_task(self._save_cursor_debounced()) - - async def _save_cursor_debounced(self) -> None: - await asyncio.sleep(CURSOR_SAVE_DEBOUNCE_S) - await self._save_session_cursors() - - async def _load_session_cursors(self) -> None: - if not self._cursor_path.exists(): - return - try: - data = json.loads(self._cursor_path.read_text("utf-8")) - except Exception as e: - logger.warning("Failed to read Mochat cursor file: {}", e) - return - cursors = data.get("cursors") if isinstance(data, dict) else None - if isinstance(cursors, dict): - for sid, cur in cursors.items(): - if isinstance(sid, str) and isinstance(cur, int) and cur >= 0: - self._session_cursor[sid] = cur - - async def _save_session_cursors(self) -> None: - try: - self._state_dir.mkdir(parents=True, exist_ok=True) - self._cursor_path.write_text(json.dumps({ - "schemaVersion": 1, "updatedAt": datetime.utcnow().isoformat(), - "cursors": self._session_cursor, - }, ensure_ascii=False, indent=2) + "\n", "utf-8") - except Exception as e: - logger.warning("Failed to save Mochat cursor file: {}", e) - - # ---- HTTP helpers ------------------------------------------------------ - - async def _post_json(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: - if not self._http: - raise RuntimeError("Mochat HTTP client not initialized") - url = f"{self.config.base_url.strip().rstrip('/')}{path}" - response = await self._http.post(url, headers={ - "Content-Type": "application/json", "X-Claw-Token": self.config.claw_token, - }, json=payload) - if not response.is_success: - raise RuntimeError(f"Mochat HTTP {response.status_code}: {response.text[:200]}") - try: - parsed = response.json() - except Exception: - parsed = response.text - if isinstance(parsed, dict) and isinstance(parsed.get("code"), int): - if parsed["code"] != 200: - msg = str(parsed.get("message") or parsed.get("name") or "request failed") - raise RuntimeError(f"Mochat API error: {msg} (code={parsed['code']})") - data = parsed.get("data") - return data if isinstance(data, dict) else {} - return parsed if isinstance(parsed, dict) else {} - - async def _api_send(self, path: str, id_key: str, id_val: str, - content: str, reply_to: str | None, group_id: str | None = None) -> dict[str, Any]: - """Unified send helper for session and panel messages.""" - body: dict[str, Any] = {id_key: id_val, "content": content} - if reply_to: - body["replyTo"] = reply_to - if group_id: - body["groupId"] = group_id - return await self._post_json(path, body) - - @staticmethod - def _read_group_id(metadata: dict[str, Any]) -> str | None: - if not isinstance(metadata, dict): - return None - value = metadata.get("group_id") or metadata.get("groupId") - return value.strip() if isinstance(value, str) and value.strip() else None diff --git a/app-instance/backend-old/nanobot/channels/qq.py b/app-instance/backend-old/nanobot/channels/qq.py deleted file mode 100644 index 5352a30..0000000 --- a/app-instance/backend-old/nanobot/channels/qq.py +++ /dev/null @@ -1,132 +0,0 @@ -"""QQ channel implementation using botpy SDK.""" - -import asyncio -from collections import deque -from typing import TYPE_CHECKING - -from loguru import logger - -from nanobot.bus.events import OutboundMessage -from nanobot.bus.queue import MessageBus -from nanobot.channels.base import BaseChannel -from nanobot.config.schema import QQConfig - -try: - import botpy - from botpy.message import C2CMessage - - QQ_AVAILABLE = True -except ImportError: - QQ_AVAILABLE = False - botpy = None - C2CMessage = None - -if TYPE_CHECKING: - from botpy.message import C2CMessage - - -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) - - class _Bot(botpy.Client): - def __init__(self): - super().__init__(intents=intents) - - async def on_ready(self): - logger.info("QQ bot ready: {}", self.robot.name) - - async def on_c2c_message_create(self, message: "C2CMessage"): - await channel._on_message(message) - - async def on_direct_message_create(self, message): - await channel._on_message(message) - - return _Bot - - -class QQChannel(BaseChannel): - """QQ channel using botpy SDK with WebSocket connection.""" - - name = "qq" - - def __init__(self, config: QQConfig, bus: MessageBus): - super().__init__(config, bus) - self.config: QQConfig = config - self._client: "botpy.Client | None" = None - self._processed_ids: deque = deque(maxlen=1000) - - async def start(self) -> None: - """Start the QQ bot.""" - if not QQ_AVAILABLE: - logger.error("QQ SDK not installed. Run: pip install qq-botpy") - return - - if not self.config.app_id or not self.config.secret: - logger.error("QQ app_id and secret not configured") - return - - self._running = True - BotClass = _make_bot_class(self) - self._client = BotClass() - - logger.info("QQ bot started (C2C private message)") - await self._run_bot() - - async def _run_bot(self) -> None: - """Run the bot connection with auto-reconnect.""" - while self._running: - try: - await self._client.start(appid=self.config.app_id, secret=self.config.secret) - except Exception as e: - logger.warning("QQ bot error: {}", e) - if self._running: - logger.info("Reconnecting QQ bot in 5 seconds...") - await asyncio.sleep(5) - - async def stop(self) -> None: - """Stop the QQ bot.""" - self._running = False - if self._client: - try: - await self._client.close() - except Exception: - pass - logger.info("QQ bot stopped") - - async def send(self, msg: OutboundMessage) -> None: - """Send a message through QQ.""" - if not self._client: - logger.warning("QQ client not initialized") - return - try: - await self._client.api.post_c2c_message( - openid=msg.chat_id, - msg_type=0, - content=msg.content, - ) - except Exception as e: - logger.error("Error sending QQ message: {}", e) - - async def _on_message(self, data: "C2CMessage") -> None: - """Handle incoming message from QQ.""" - try: - # Dedup by message ID - if data.id in self._processed_ids: - return - self._processed_ids.append(data.id) - - author = data.author - user_id = str(getattr(author, 'id', None) or getattr(author, 'user_openid', 'unknown')) - content = (data.content or "").strip() - if not content: - return - - await self._handle_message( - sender_id=user_id, - chat_id=user_id, - content=content, - metadata={"message_id": data.id}, - ) - except Exception: - logger.exception("Error handling QQ message") diff --git a/app-instance/backend-old/nanobot/channels/slack.py b/app-instance/backend-old/nanobot/channels/slack.py deleted file mode 100644 index 906593b..0000000 --- a/app-instance/backend-old/nanobot/channels/slack.py +++ /dev/null @@ -1,257 +0,0 @@ -"""Slack channel implementation using Socket Mode.""" - -import asyncio -import re -from typing import Any - -from loguru import logger -from slack_sdk.socket_mode.websockets import SocketModeClient -from slack_sdk.socket_mode.request import SocketModeRequest -from slack_sdk.socket_mode.response import SocketModeResponse -from slack_sdk.web.async_client import AsyncWebClient - -from slackify_markdown import slackify_markdown - -from nanobot.bus.events import OutboundMessage -from nanobot.bus.queue import MessageBus -from nanobot.channels.base import BaseChannel -from nanobot.config.schema import SlackConfig - - -class SlackChannel(BaseChannel): - """Slack channel using Socket Mode.""" - - name = "slack" - - def __init__(self, config: SlackConfig, bus: MessageBus): - super().__init__(config, bus) - self.config: SlackConfig = config - self._web_client: AsyncWebClient | None = None - self._socket_client: SocketModeClient | None = None - self._bot_user_id: str | None = None - - async def start(self) -> None: - """Start the Slack Socket Mode client.""" - if not self.config.bot_token or not self.config.app_token: - logger.error("Slack bot/app token not configured") - return - if self.config.mode != "socket": - logger.error("Unsupported Slack mode: {}", self.config.mode) - return - - self._running = True - - self._web_client = AsyncWebClient(token=self.config.bot_token) - self._socket_client = SocketModeClient( - app_token=self.config.app_token, - web_client=self._web_client, - ) - - self._socket_client.socket_mode_request_listeners.append(self._on_socket_request) - - # Resolve bot user ID for mention handling - try: - auth = await self._web_client.auth_test() - self._bot_user_id = auth.get("user_id") - logger.info("Slack bot connected as {}", self._bot_user_id) - except Exception as e: - logger.warning("Slack auth_test failed: {}", e) - - logger.info("Starting Slack Socket Mode client...") - await self._socket_client.connect() - - while self._running: - await asyncio.sleep(1) - - async def stop(self) -> None: - """Stop the Slack client.""" - self._running = False - if self._socket_client: - try: - await self._socket_client.close() - except Exception as e: - logger.warning("Slack socket close failed: {}", e) - self._socket_client = None - - async def send(self, msg: OutboundMessage) -> None: - """Send a message through Slack.""" - if not self._web_client: - logger.warning("Slack client not running") - return - try: - slack_meta = msg.metadata.get("slack", {}) if msg.metadata else {} - thread_ts = slack_meta.get("thread_ts") - channel_type = slack_meta.get("channel_type") - # Only reply in thread for channel/group messages; DMs don't use threads - use_thread = thread_ts and channel_type != "im" - thread_ts_param = thread_ts if use_thread else None - - if msg.content: - await self._web_client.chat_postMessage( - channel=msg.chat_id, - text=self._to_mrkdwn(msg.content), - thread_ts=thread_ts_param, - ) - - for media_path in msg.media or []: - try: - await self._web_client.files_upload_v2( - channel=msg.chat_id, - file=media_path, - thread_ts=thread_ts_param, - ) - except Exception as e: - logger.error("Failed to upload file {}: {}", media_path, e) - except Exception as e: - logger.error("Error sending Slack message: {}", e) - - async def _on_socket_request( - self, - client: SocketModeClient, - req: SocketModeRequest, - ) -> None: - """Handle incoming Socket Mode requests.""" - if req.type != "events_api": - return - - # Acknowledge right away - await client.send_socket_mode_response( - SocketModeResponse(envelope_id=req.envelope_id) - ) - - payload = req.payload or {} - event = payload.get("event") or {} - event_type = event.get("type") - - # Handle app mentions or plain messages - if event_type not in ("message", "app_mention"): - return - - sender_id = event.get("user") - chat_id = event.get("channel") - - # Ignore bot/system messages (any subtype = not a normal user message) - if event.get("subtype"): - return - if self._bot_user_id and sender_id == self._bot_user_id: - return - - # Avoid double-processing: Slack sends both `message` and `app_mention` - # for mentions in channels. Prefer `app_mention`. - text = event.get("text") or "" - if event_type == "message" and self._bot_user_id and f"<@{self._bot_user_id}>" in text: - return - - # Debug: log basic event shape - logger.debug( - "Slack event: type={} subtype={} user={} channel={} channel_type={} text={}", - event_type, - event.get("subtype"), - sender_id, - chat_id, - event.get("channel_type"), - text[:80], - ) - if not sender_id or not chat_id: - return - - channel_type = event.get("channel_type") or "" - - if not self._is_allowed(sender_id, chat_id, channel_type): - return - - if channel_type != "im" and not self._should_respond_in_channel(event_type, text, chat_id): - return - - text = self._strip_bot_mention(text) - - thread_ts = event.get("thread_ts") - if self.config.reply_in_thread and not thread_ts: - thread_ts = event.get("ts") - # Add :eyes: reaction to the triggering message (best-effort) - try: - if self._web_client and event.get("ts"): - await self._web_client.reactions_add( - channel=chat_id, - name=self.config.react_emoji, - timestamp=event.get("ts"), - ) - except Exception as e: - logger.debug("Slack reactions_add failed: {}", e) - - # Thread-scoped session key for channel/group messages - session_key = f"slack:{chat_id}:{thread_ts}" if thread_ts and channel_type != "im" else None - - try: - await self._handle_message( - sender_id=sender_id, - chat_id=chat_id, - content=text, - metadata={ - "slack": { - "event": event, - "thread_ts": thread_ts, - "channel_type": channel_type, - }, - }, - session_key=session_key, - ) - except Exception: - logger.exception("Error handling Slack message from {}", sender_id) - - def _is_allowed(self, sender_id: str, chat_id: str, channel_type: str) -> bool: - if channel_type == "im": - if not self.config.dm.enabled: - return False - if self.config.dm.policy == "allowlist": - return sender_id in self.config.dm.allow_from - return True - - # Group / channel messages - if self.config.group_policy == "allowlist": - return chat_id in self.config.group_allow_from - return True - - def _should_respond_in_channel(self, event_type: str, text: str, chat_id: str) -> bool: - if self.config.group_policy == "open": - return True - if self.config.group_policy == "mention": - if event_type == "app_mention": - return True - return self._bot_user_id is not None and f"<@{self._bot_user_id}>" in text - if self.config.group_policy == "allowlist": - return chat_id in self.config.group_allow_from - return False - - def _strip_bot_mention(self, text: str) -> str: - if not text or not self._bot_user_id: - return text - return re.sub(rf"<@{re.escape(self._bot_user_id)}>\s*", "", text).strip() - - _TABLE_RE = re.compile(r"(?m)^\|.*\|$(?:\n\|[\s:|-]*\|$)(?:\n\|.*\|$)*") - - @classmethod - def _to_mrkdwn(cls, text: str) -> str: - """Convert Markdown to Slack mrkdwn, including tables.""" - if not text: - return "" - text = cls._TABLE_RE.sub(cls._convert_table, text) - return slackify_markdown(text) - - @staticmethod - def _convert_table(match: re.Match) -> str: - """Convert a Markdown table to a Slack-readable list.""" - lines = [ln.strip() for ln in match.group(0).strip().splitlines() if ln.strip()] - if len(lines) < 2: - return match.group(0) - headers = [h.strip() for h in lines[0].strip("|").split("|")] - start = 2 if re.fullmatch(r"[|\s:\-]+", lines[1]) else 1 - rows: list[str] = [] - for line in lines[start:]: - cells = [c.strip() for c in line.strip("|").split("|")] - cells = (cells + [""] * len(headers))[: len(headers)] - parts = [f"**{headers[i]}**: {cells[i]}" for i in range(len(headers)) if cells[i]] - if parts: - rows.append(" · ".join(parts)) - return "\n".join(rows) - diff --git a/app-instance/backend-old/nanobot/channels/telegram.py b/app-instance/backend-old/nanobot/channels/telegram.py deleted file mode 100644 index 3a672e0..0000000 --- a/app-instance/backend-old/nanobot/channels/telegram.py +++ /dev/null @@ -1,457 +0,0 @@ -"""Telegram channel implementation using python-telegram-bot.""" - -from __future__ import annotations - -import asyncio -import re -from loguru import logger -from telegram import BotCommand, Update, ReplyParameters -from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes -from telegram.request import HTTPXRequest - -from nanobot.bus.events import OutboundMessage -from nanobot.bus.queue import MessageBus -from nanobot.channels.base import BaseChannel -from nanobot.config.schema import TelegramConfig - - -def _markdown_to_telegram_html(text: str) -> str: - """ - Convert markdown to Telegram-safe HTML. - """ - if not text: - return "" - - # 1. Extract and protect code blocks (preserve content from other processing) - code_blocks: list[str] = [] - def save_code_block(m: re.Match) -> str: - code_blocks.append(m.group(1)) - return f"\x00CB{len(code_blocks) - 1}\x00" - - text = re.sub(r'```[\w]*\n?([\s\S]*?)```', save_code_block, text) - - # 2. Extract and protect inline code - inline_codes: list[str] = [] - def save_inline_code(m: re.Match) -> str: - inline_codes.append(m.group(1)) - return f"\x00IC{len(inline_codes) - 1}\x00" - - text = re.sub(r'`([^`]+)`', save_inline_code, text) - - # 3. Headers # Title -> just the title text - text = re.sub(r'^#{1,6}\s+(.+)$', r'\1', text, flags=re.MULTILINE) - - # 4. Blockquotes > text -> just the text (before HTML escaping) - text = re.sub(r'^>\s*(.*)$', r'\1', text, flags=re.MULTILINE) - - # 5. Escape HTML special characters - text = text.replace("&", "&").replace("<", "<").replace(">", ">") - - # 6. Links [text](url) - must be before bold/italic to handle nested cases - text = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', r'\1', text) - - # 7. Bold **text** or __text__ - text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) - text = re.sub(r'__(.+?)__', r'\1', text) - - # 8. Italic _text_ (avoid matching inside words like some_var_name) - text = re.sub(r'(?\1', text) - - # 9. Strikethrough ~~text~~ - text = re.sub(r'~~(.+?)~~', r'{escaped}")
-
- # 12. Restore code blocks with HTML tags
- for i, code in enumerate(code_blocks):
- # Escape HTML in code content
- escaped = code.replace("&", "&").replace("<", "<").replace(">", ">")
- text = text.replace(f"\x00CB{i}\x00", f"{escaped}
- {pickAppText(locale, '已安装位置：全局插件目录或当前 workspace 的 ', 'Installed from the global plugin directory or this workspace\'s ')}
- plugins/
-
{pickAppText(locale, '还没有安装任何插件', 'No plugins are installed yet')}
-
- {pickAppText(locale, '把插件目录放到全局插件目录或当前 workspace 的 ', 'Put a plugin directory in the global plugin directory or this workspace\'s ')}
- plugins/
- {pickAppText(locale, '，然后重启 Boardware Agent Sandbox。', ', then restart Boardware Agent Sandbox.')}
-
- {plugin.description} -
- )} -
- {agent.name}
-
- {agent.description || '—'}
-/{cmd.name}
- {cmd.argument_hint && (
- {cmd.argument_hint}
- )}
- - {cmd.description || '—'} -
-{pickAppText(locale, '智能体、A2A、MCP 的实时过程', 'Live process stream for agents, A2A, and MCP')}
-{run.title}
-