feat(onebot): support main-port WebSocket transports and filter protocol events

skills/scripts/e2e/onebot-runtime-probe.py

@@ -6,14 +6,15 @@ from __future__ import annotations
 import argparse
 import asyncio
 import json
+import os
 import time
 
 import websockets
 
 
-async def run(port: int) -> None:
-    uri = f'ws://127.0.0.1:{port}/ws'
+async def run(uri: str) -> None:
     headers = {
+        'Authorization': 'Bearer ' + os.environ['ONEBOT_PROBE_TOKEN'],
         'X-Self-ID': '900001',
         'X-Client-Role': 'Universal',
         'User-Agent': 'LangBot-E2E-OneBot/1.0',
@@ -27,7 +28,7 @@ async def run(port: int) -> None:
         except OSError:
             await asyncio.sleep(0.25)
     if connection is None:
-        raise RuntimeError(f'OneBot reverse WebSocket did not open on port {port}')
+        raise RuntimeError('OneBot reverse WebSocket did not open')
 
     actions: list[str] = []
     event_deadline = time.monotonic() + 20
@@ -88,6 +89,6 @@ async def run(port: int) -> None:
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser()
-    parser.add_argument('--port', type=int, required=True)
+    parser.add_argument('--url', required=True)
     args = parser.parse_args()
-    asyncio.run(run(args.port))
+    asyncio.run(run(args.url))

skills/scripts/e2e/wizard-onebot-agent-runtime.mjs

@@ -1,7 +1,7 @@
 #!/usr/bin/env node
 
 import { spawn } from "node:child_process";
-import { createServer } from "node:net";
+import { randomBytes } from "node:crypto";
 import { dirname, join } from "node:path";
 import { fileURLToPath } from "node:url";
 import {
@@ -60,19 +60,6 @@ const result = {
   evidence_collected: ["ui", "screenshot", "console", "api_diagnostic"],
 };
 
-async function getFreePort() {
-  const server = createServer();
-  await new Promise((resolve, reject) => {
-    server.once("error", reject);
-    server.listen(0, "127.0.0.1", resolve);
-  });
-  const address = server.address();
-  const port = typeof address === "object" && address ? address.port : 0;
-  await new Promise((resolve) => server.close(resolve));
-  if (!port) throw new Error("Could not allocate a temporary OneBot port.");
-  return port;
-}
-
 async function api(page, path, options = {}) {
   return await page.evaluate(
     async ({ baseUrl, path, options, token }) => {
@@ -96,7 +83,7 @@ async function api(page, path, options = {}) {
   );
 }
 
-async function runProbe(port) {
+async function runProbe(url, accessToken) {
   const repo = await resolveLangBotRepo();
   const python = join(repo, ".venv", "bin", "python");
   const script = join(
@@ -107,8 +94,9 @@ async function runProbe(port) {
     throw new Error(`LangBot virtualenv Python is missing: ${python}`);
   }
   return await new Promise((resolve, reject) => {
-    const child = spawn(python, [script, "--port", String(port)], {
+    const child = spawn(python, [script, "--url", url], {
       cwd: repo,
+      env: { ...process.env, ONEBOT_PROBE_TOKEN: accessToken },
     });
     let stdout = "";
     let stderr = "";
@@ -132,7 +120,7 @@ async function runProbe(port) {
 try {
   if (!frontendUrl) throw new Error("LANGBOT_FRONTEND_URL is not configured.");
   if (!backendUrl) throw new Error("LANGBOT_BACKEND_URL is not configured.");
-  const port = await getFreePort();
+  const accessToken = randomBytes(32).toString("hex");
 
   browser = await createBrowser(paths);
   const { page } = browser;
@@ -190,9 +178,7 @@ try {
     .getByText(/Configure Your Bot|配置机器人|ボットを設定/)
     .first()
     .waitFor({ timeout: 15_000 });
-  const portInput = page.getByRole("spinbutton").first();
-  await portInput.fill(String(port));
-  await portInput.press("Tab");
+  await page.getByLabel(/Access token|访问令牌|アクセストークン/i).fill(accessToken);
   await page
     .getByRole("button", {
       name: /Save & Enable Bot|保存并启用|保存して有効化/,
@@ -270,7 +256,7 @@ try {
   }
   result.visible_signals.push("route-persisted");
 
-  result.runtime.onebot = await runProbe(port);
+  result.runtime.onebot = await runProbe(`${backendUrl.replace(/^http/, "ws").replace(/\/$/, "")}/bots/${botId}/onebot/v11/ws`, accessToken);
   let routeStatus;
   for (let attempt = 0; attempt < 30; attempt += 1) {
     routeStatus = await api(

skills/skills/langbot-testing/cases/wizard-onebot-agent-runtime.yaml

@@ -32,7 +32,7 @@ preconditions:
   - "The LangBot virtualenv includes the websockets package used by the local OneBot probe."
 steps:
   - "Open Quick Start and choose Welcome new members with the OneBot v11 channel."
-  - "Create and enable a temporary OneBot Bot on an available local port."
+  - "Create and enable a temporary OneBot Bot using the main-port reverse WS URL and a generated access token."
   - "Select Local Agent and confirm the editable welcome prompt is prefilled."
   - "Choose QA Deterministic Runner and finish the wizard."
   - "Connect a local OneBot reverse WebSocket and send a group_increase notice."

src/langbot/pkg/api/http/controller/groups/webhooks.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import quart
+import quart_cors
 import traceback
 
 from .. import group
@@ -10,6 +11,35 @@ from .....utils import bounded_executor
 @group.group_class('webhooks', '/bots')
 class WebhookRouterGroup(group.RouterGroup):
     async def initialize(self) -> None:
+        @self.quart_app.websocket(self.path + '/<bot_uuid>/onebot/v11/ws')
+        @quart_cors.cors_exempt
+        async def onebot_websocket(bot_uuid: str):
+            # Protocol clients omit Origin; bearer authentication protects this endpoint.
+            from .....platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter
+
+            runtime = await self.ap.platform_mgr.resolve_public_bot(bot_uuid)
+            if runtime is None or not runtime.enable or not isinstance(runtime.adapter, AiocqhttpAdapter):
+                quart.abort(404)
+            adapter = runtime.adapter
+            socket = quart.websocket._get_current_object()
+            if adapter.config['connection_mode'] != 'reverse_ws':
+                quart.abort(409)
+            if not adapter.authenticate(socket.headers):
+                quart.abort(401)
+            if socket.headers.get('X-Client-Role', '').lower() != 'universal':
+                quart.abort(400)
+            self_id = socket.headers.get('X-Self-ID', '')
+            expected = str(adapter.config.get('self_id', '') or '')
+            if not self_id or (expected and self_id != expected):
+                quart.abort(400)
+            if adapter.bot.connection is not None:
+                quart.abort(409)
+            await socket.accept()
+            try:
+                await adapter.serve_connection(socket.send, socket.receive, socket.close, self_id)
+            except Exception:
+                await socket.close(1011)
+
         @self.route('/<bot_uuid>', methods=['GET', 'POST'], auth_type=group.AuthType.NONE)
         async def handle_webhook(bot_uuid: str):
             """处理 bot webhook 回调（无子路径）"""

src/langbot/pkg/api/http/service/bot.py

@@ -52,13 +52,13 @@ class BotService:
         return None
 
     def _adapter_declares_webhook_url(self, adapter_name: str) -> bool:
-        """Whether the adapter manifest declares a generated webhook URL config item."""
+        """Whether the manifest needs a generated public bot ingress URL."""
         component = self._get_adapter_component(adapter_name)
         if component is None:
             return False
 
         for config_item in component.spec.get('config', []):
-            if config_item.get('type') == 'webhook-url':
+            if config_item.get('type') == 'webhook-url' or config_item.get('name') == '__system.onebot_ws_url':
                 return True
         return False
 

src/langbot/pkg/platform/adapter_names.py

@@ -2,7 +2,6 @@
 
 OMNI_ADAPTER_NAMES = frozenset(
     {
-        'aiocqhttp',
         'telegram',
         'discord',
         'qqofficial',
@@ -20,6 +19,8 @@ OMNI_ADAPTER_NAMES = frozenset(
 
 def canonical_adapter_name(name: str) -> str:
     """Accept previous built-in IDs without renaming legacy or custom adapters."""
+    if name in ('aiocqhttp-omni', 'aiocqhttp-eba'):
+        return 'onebot-v11'
     if name.endswith('-eba') and name[:-4] in OMNI_ADAPTER_NAMES:
         return f'{name[:-4]}-omni'
     return name

src/langbot/pkg/platform/adapters/aiocqhttp/adapter.py

@@ -2,6 +2,12 @@ from __future__ import annotations
 
 import asyncio
 import traceback
+import hmac
+from urllib.parse import urlsplit
+
+from websockets.asyncio.client import connect
+from langbot.pkg.core.task_boundary import create_detached_task
+from langbot.pkg.platform.adapters.aiocqhttp.transport import OneBotClient, OneBotConnection
 import typing
 
 import aiocqhttp
@@ -20,7 +26,10 @@ from langbot_plugin.api.entities.builtin.platform.errors import NotSupportedErro
 
 
 class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlatformAdapter):
-    bot: aiocqhttp.CQHttp = pydantic.Field(exclude=True)
+    bot: typing.Any = pydantic.Field(exclude=True)
+    _runtime: typing.Any = pydantic.PrivateAttr(default=None)
+    _stopped: asyncio.Event = pydantic.PrivateAttr(default_factory=asyncio.Event)
+    _connection_task: typing.Any = pydantic.PrivateAttr(default=None)
 
     message_converter: AiocqhttpMessageConverter = AiocqhttpMessageConverter()
     event_converter: AiocqhttpEventConverter = AiocqhttpEventConverter()
@@ -36,24 +45,64 @@ class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlat
         arbitrary_types_allowed = True
 
     def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger):
-        run_config = dict(config)
+        config = dict(config)
+        config.setdefault('connection_mode', 'reverse_ws')
+        if not isinstance(config.get('access-token'), str) or not config['access-token'].strip():
+            raise ValueError('OneBot requires a non-empty access token')
+        if config['connection_mode'] not in ('reverse_ws', 'forward_ws'):
+            raise ValueError('Unsupported OneBot connection mode')
+        if config['connection_mode'] == 'forward_ws':
+            url = urlsplit(config.get('ws_url', ''))
+            if url.scheme not in ('ws', 'wss') or not url.hostname or url.username or url.password or url.fragment:
+                raise ValueError('OneBot ws_url must be a ws:// or wss:// endpoint without embedded credentials')
+        super().__init__(config=config, logger=logger, bot=OneBotClient(), bot_account_id='', listeners={})
 
-        async def shutdown_trigger_placeholder():
-            while True:
-                await asyncio.sleep(1)
+    def bind_runtime(self, runtime):
+        self._runtime = runtime
 
-        run_config['shutdown_trigger'] = shutdown_trigger_placeholder
-        access_token = run_config.pop('access-token', '') or None
-        bot = aiocqhttp.CQHttp(access_token=access_token)
+    async def _validate_connection(self):
+        if self._stopped.is_set() or self._runtime is None:
+            raise RuntimeError('OneBot runtime is inactive')
+        current = await self._runtime.ap.platform_mgr.resolve_public_bot(self._runtime.bot_entity.uuid)
+        if current is not self._runtime or not current.enable:
+            raise RuntimeError('OneBot runtime placement changed')
 
-        super().__init__(
-            config=run_config,
-            logger=logger,
-            bot=bot,
-            bot_account_id='',
-            listeners={},
+    def authenticate(self, headers):
+        token = str(self.config.get('access-token', ''))
+        supplied = headers.get('Authorization', '')
+        return bool(token) and hmac.compare_digest(supplied.encode(), ('Bearer ' + token).encode())
+
+    async def serve_connection(self, send, receive, close, self_id=''):
+        await self._validate_connection()
+        if self.bot.connection is not None:
+            raise RuntimeError('OneBot already has an active connection')
+        expected = str(self.config.get('self_id', '') or '')
+        if expected and self_id and expected != self_id:
+            raise ValueError('OneBot account mismatch')
+        connection = OneBotConnection(
+            send,
+            receive,
+            close,
+            self._receive_event,
+            self._validate_connection,
+            self_id=expected or self_id,
         )
-        self._register_native_handlers()
+        self.bot.connection = connection
+        self.bot_account_id = expected or self_id
+        self._lookup.clear()
+        task = create_detached_task(connection.run(), workspace_uuid=self._runtime.workspace_uuid)
+        self._connection_task = task
+        try:
+            await task
+        finally:
+            await connection.close()
+            if self.bot.connection is connection:
+                self.bot.connection = None
+                self._connection_task = None
+                self._lookup.clear()
+
+    async def _receive_event(self, payload):
+        await self._handle_native_event(aiocqhttp.Event.from_payload(payload))
 
     def get_supported_events(self) -> list[str]:
         return [
@@ -121,33 +170,51 @@ class AiocqhttpAdapter(AiocqhttpAPIMixin, abstract_platform_adapter.AbstractPlat
             self.listeners.pop(event_type, None)
 
     async def run_async(self):
-        await self.bot._server_app.run_task(**self.config)
+        if not self.config.get('access-token'):
+            raise ValueError('OneBot requires a non-empty access token')
+        if self.config['connection_mode'] == 'reverse_ws':
+            await self._stopped.wait()
+            return
+        delay = 1
+        while not self._stopped.is_set():
+            await self._validate_connection()
+            try:
+                async with connect(
+                    self.config['ws_url'],
+                    additional_headers={'Authorization': 'Bearer ' + self.config['access-token']},
+                    open_timeout=10,
+                    close_timeout=5,
+                    ping_interval=20,
+                    ping_timeout=20,
+                    max_size=8 * 1024 * 1024,
+                    proxy=None,
+                ) as socket:
+                    delay = 1
+                    await self.serve_connection(socket.send, socket.recv, socket.close)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                await self.logger.warning(f'OneBot connection lost ({type(exc).__name__}); reconnecting')
+            try:
+                await asyncio.wait_for(self._stopped.wait(), timeout=delay)
+            except TimeoutError:
+                pass
+            delay = min(delay * 2, 30)
 
     async def kill(self) -> bool:
+        self._stopped.set()
+        task = self._connection_task
+        if task is not None:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
         self._lookup.clear()
-        return False
-
-    def _register_native_handlers(self):
-        @self.bot.on_message()
-        async def on_message(event: aiocqhttp.Event):
-            await self._handle_native_event(event)
-
-        @self.bot.on_notice()
-        async def on_notice(event: aiocqhttp.Event):
-            await self._handle_native_event(event)
-
-        @self.bot.on_request()
-        async def on_request(event: aiocqhttp.Event):
-            await self._handle_native_event(event)
-
-        @self.bot.on_websocket_connection
-        async def on_websocket_connection(event: aiocqhttp.Event):
-            self.bot_account_id = str(getattr(event, 'self_id', '') or self.bot_account_id)
-            await self.logger.info(f'WebSocket connection established, bot id: {self.bot_account_id}')
-            await self._dispatch_native_event(event)
+        return True
 
     async def _handle_native_event(self, event: aiocqhttp.Event):
         self.bot_account_id = str(getattr(event, 'self_id', '') or self.bot_account_id)
+        # Connection metadata belongs to the adapter, never business event routing.
+        if getattr(event, 'type', None) == 'meta_event':
+            return
         if getattr(event, 'type', None) == 'message' and str(getattr(event, 'user_id', '')) == self.bot_account_id:
             return
         try:

src/langbot/pkg/platform/adapters/aiocqhttp/event_converter.py

@@ -34,8 +34,6 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
             return AiocqhttpEventConverter.notice_to_eba(event, bot_user_id)
         if event_type == 'request':
             return AiocqhttpEventConverter.request_to_eba(event)
-        if event_type == 'meta_event':
-            return AiocqhttpEventConverter.platform_specific(event, f'meta.{getattr(event, "detail_type", "")}')
         return None
 
     @staticmethod
@@ -90,7 +88,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
             )
         return platform_events.MessageReceivedEvent(
             type='message.received',
-            adapter_name='aiocqhttp-omni',
+            adapter_name='onebot-v11',
             message_id=getattr(event, 'message_id', ''),
             message_chain=message_chain,
             sender=sender,
@@ -111,7 +109,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
         if notice_type in ('group_recall', 'friend_recall'):
             return platform_events.MessageDeletedEvent(
                 type='message.deleted',
-                adapter_name='aiocqhttp-omni',
+                adapter_name='onebot-v11',
                 message_id=getattr(event, 'message_id', ''),
                 operator=AiocqhttpEventConverter.user(getattr(event, 'operator_id', None)),
                 chat_type=platform_entities.ChatType.GROUP
@@ -129,7 +127,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
             if AiocqhttpEventConverter._is_bot_user(getattr(event, 'user_id', None), bot_user_id, event):
                 return platform_events.BotInvitedToGroupEvent(
                     type='bot.invited_to_group',
-                    adapter_name='aiocqhttp-omni',
+                    adapter_name='onebot-v11',
                     group=group,
                     inviter=AiocqhttpEventConverter.user(inviter_id) if inviter_id else None,
                     timestamp=float(getattr(event, 'time', 0) or 0),
@@ -137,7 +135,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
                 )
             return platform_events.MemberJoinedEvent(
                 type='group.member_joined',
-                adapter_name='aiocqhttp-omni',
+                adapter_name='onebot-v11',
                 group=group,
                 member=user,
                 inviter=AiocqhttpEventConverter.user(inviter_id) if inviter_id else None,
@@ -151,7 +149,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
             if AiocqhttpEventConverter._is_bot_user(getattr(event, 'user_id', None), bot_user_id, event):
                 return platform_events.BotRemovedFromGroupEvent(
                     type='bot.removed_from_group',
-                    adapter_name='aiocqhttp-omni',
+                    adapter_name='onebot-v11',
                     group=group,
                     operator=operator,
                     timestamp=float(getattr(event, 'time', 0) or 0),
@@ -159,7 +157,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
                 )
             return platform_events.MemberLeftEvent(
                 type='group.member_left',
-                adapter_name='aiocqhttp-omni',
+                adapter_name='onebot-v11',
                 group=group,
                 member=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')),
                 is_kicked=getattr(event, 'sub_type', '') in ('kick', 'kick_me'),
@@ -175,7 +173,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
                 event_cls = platform_events.BotMutedEvent if duration > 0 else platform_events.BotUnmutedEvent
                 kwargs: dict[str, typing.Any] = {
                     'type': 'bot.muted' if duration > 0 else 'bot.unmuted',
-                    'adapter_name': 'aiocqhttp-omni',
+                    'adapter_name': 'onebot-v11',
                     'group': group,
                     'operator': operator,
                     'timestamp': float(getattr(event, 'time', 0) or 0),
@@ -187,7 +185,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
             if duration > 0:
                 return platform_events.MemberBannedEvent(
                     type='group.member_banned',
-                    adapter_name='aiocqhttp-omni',
+                    adapter_name='onebot-v11',
                     group=group,
                     member=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')),
                     operator=operator,
@@ -198,7 +196,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
         if notice_type == 'friend_add':
             return platform_events.FriendAddedEvent(
                 type='friend.added',
-                adapter_name='aiocqhttp-omni',
+                adapter_name='onebot-v11',
                 user=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')),
                 timestamp=float(getattr(event, 'time', 0) or 0),
                 source_platform_object=event,
@@ -211,7 +209,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
         if request_type == 'friend':
             return platform_events.FriendRequestReceivedEvent(
                 type='friend.request_received',
-                adapter_name='aiocqhttp-omni',
+                adapter_name='onebot-v11',
                 request_id=getattr(event, 'flag', ''),
                 user=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')),
                 message=getattr(event, 'comment', None),
@@ -221,7 +219,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
         if request_type == 'group' and getattr(event, 'sub_type', '') == 'invite':
             return platform_events.BotInvitedToGroupEvent(
                 type='bot.invited_to_group',
-                adapter_name='aiocqhttp-omni',
+                adapter_name='onebot-v11',
                 group=AiocqhttpEventConverter.group_from_event(event),
                 inviter=AiocqhttpEventConverter.user(getattr(event, 'user_id', '')),
                 request_id=getattr(event, 'flag', ''),
@@ -258,7 +256,7 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
     def platform_specific(event: aiocqhttp.Event, action: str) -> platform_events.PlatformSpecificEvent:
         return platform_events.PlatformSpecificEvent(
             type='platform.specific',
-            adapter_name='aiocqhttp-omni',
+            adapter_name='onebot-v11',
             action=action,
             data={key: value for key, value in dict(event).items() if key not in {'message'}},
             timestamp=float(getattr(event, 'time', 0) or 0),

src/langbot/pkg/platform/adapters/aiocqhttp/manifest.yaml

@@ -2,7 +2,7 @@ apiVersion: v1
 kind: MessagePlatformAdapter
 
 metadata:
-  name: aiocqhttp-omni
+  name: onebot-v11
   label:
     en_US: OneBot v11
     zh_Hans: OneBot v11
@@ -22,42 +22,59 @@ spec:
     en: https://langbot.app/docs/en/usage/platforms/qq/aiocqhttp/napcat
     ja: https://langbot.app/docs/ja/usage/platforms/qq/aiocqhttp/napcat
   config:
-    - name: host
-      label:
-        en_US: Host
-        zh_Hans: 主机
-        zh_Hant: 主機
-      description:
-        en_US: The host that OneBot v11 listens on for reverse WebSocket connections. Unless you know what you're doing, use 0.0.0.0
-        zh_Hans: OneBot v11 反向 WebSocket 监听主机，除非你知道自己在做什么，否则请写 0.0.0.0
-        zh_Hant: OneBot v11 反向 WebSocket 監聽主機，除非你知道自己在做什麼，否則請填 0.0.0.0
-      type: string
+    - name: connection_mode
+      label: {en_US: Connection mode, zh_Hans: 连接模式, zh_Hant: 連線模式, ja_JP: 接続方式}
+      type: select
       required: true
-      default: 0.0.0.0
-    - name: port
-      label:
-        en_US: Port
-        zh_Hans: 端口
-        zh_Hant: 連接埠
+      default: reverse_ws
+      options:
+        - name: reverse_ws
+          label: {en_US: 'Reverse WS — protocol endpoint connects to LangBot', zh_Hans: '反向 WS — 协议端连接 LangBot', zh_Hant: '反向 WS — 協定端連接 LangBot', ja_JP: '逆方向 WS — プロトコル側から LangBot に接続'}
+        - name: forward_ws
+          label: {en_US: 'Forward WS — LangBot connects to protocol endpoint', zh_Hans: '正向 WS — LangBot 连接协议端', zh_Hant: '正向 WS — LangBot 連接協定端', ja_JP: '順方向 WS — LangBot からプロトコル側に接続'}
       description:
-        en_US: Reverse WebSocket listen port
-        zh_Hans: 反向 WebSocket 监听端口
-        zh_Hant: 反向 WebSocket 監聽連接埠
-      type: integer
-      required: true
-      default: 2280
-    - name: access-token
-      label:
-        en_US: Access Token
-        zh_Hans: 访问令牌
-        zh_Hant: 存取令牌
+        en_US: Reverse WS uses LangBot's main service port. Forward WS requires a protocol endpoint reachable from LangBot. HTTP is not supported.
+        zh_Hans: 反向 WS 由协议端连接 LangBot 主服务端口，无需额外监听端口；正向 WS 由 LangBot 连接可访问的协议端。不支持 HTTP。
+        zh_Hant: 反向 WS 由協定端連接 LangBot 主服務連接埠；正向 WS 由 LangBot 連接可存取的協定端。不支援 HTTP。
+        ja_JP: 逆方向 WS は LangBot のメインポートを使用。順方向 WS は LangBot から到達可能なプロトコル側に接続します。HTTP は非対応です。
+    - name: __system.onebot_ws_url
+      label: {en_US: Protocol endpoint connection URL, zh_Hans: 协议端连接地址, zh_Hant: 協定端連線位址, ja_JP: プロトコル側の接続先 URL}
       description:
-        en_US: Custom connection token for the protocol endpoint. Leave empty if the endpoint has no token configured
-        zh_Hans: 自定义的协议端连接令牌；若协议端未设置，则不填
-        zh_Hant: 自訂的協定端連線令牌；若協定端未設定，則不填
+        en_US: Save the bot, then copy this URL into the protocol endpoint's WebSocket client (Universal). Old host/port settings are ignored; update the protocol endpoint URL.
+        zh_Hans: 保存机器人后，将此地址填入协议端的 WebSocket 客户端（Universal）配置。旧 host/port 配置不再生效，请同步修改协议端地址。
+        zh_Hant: 儲存後將此位址填入協定端 WebSocket 用戶端（Universal）。舊 host/port 不再生效，請同步更新協定端位址。
+        ja_JP: 保存後、この URL をプロトコル側の WebSocket クライアント（Universal）に設定。旧 host/port は無効です。接続先を更新してください。
       type: string
       required: false
-      default: ""
+      default: ''
+      show_if: {field: connection_mode, operator: eq, value: reverse_ws}
+    - name: ws_url
+      label: {en_US: Protocol endpoint WebSocket URL, zh_Hans: 协议端 WebSocket 地址, zh_Hant: 協定端 WebSocket 位址, ja_JP: プロトコル側 WebSocket URL}
+      description:
+        en_US: LangBot connects to this ws:// or wss:// Universal endpoint. It must be reachable from LangBot.
+        zh_Hans: LangBot 将主动连接此 ws:// 或 wss:// 地址，使用协议端同时提供事件和 API 的 Universal 端点。请确保 LangBot 能访问它。
+        zh_Hant: LangBot 主動連接此 ws:// 或 wss:// Universal 位址，請確保可存取。
+        ja_JP: LangBot がこの ws:// または wss:// Universal URL に接続します。到達可能な URL を指定してください。
+      type: string
+      required: true
+      default: ''
+      show_if: {field: connection_mode, operator: eq, value: forward_ws}
+    - name: access-token
+      label: {en_US: Access token, zh_Hans: 访问令牌, zh_Hant: 存取權杖, ja_JP: アクセストークン}
+      description:
+        en_US: Required in both modes. Configure the same secret token on LangBot and the protocol endpoint. Sent using Authorization Bearer.
+        zh_Hans: 两种模式均必填。在 LangBot 与协议端设置相同的保密令牌，连接时使用 Authorization Bearer 验证。
+        zh_Hant: 兩種模式皆必填。在 LangBot 與協定端設定相同的保密權杖。
+        ja_JP: 両方式で必須。LangBot とプロトコル側に同じ秘密トークンを設定してください。
+      type: string
+      required: true
+      default: ''
+    - name: self_id
+      label: {en_US: Expected bot account (optional), zh_Hans: 预期机器人账号（可选）, zh_Hant: 預期機器人帳號（選填）, ja_JP: ボットアカウント（任意）}
+      description: {en_US: Reject connections or events from a different account., zh_Hans: 拒绝其他账号的连接或事件。, zh_Hant: 拒絕其他帳號的連線或事件。, ja_JP: 異なるアカウントの接続やイベントを拒否します。}
+      type: string
+      required: false
+      default: ''
 
   supported_events:
     - message.received

src/langbot/pkg/platform/adapters/aiocqhttp/transport.py

@@ -0,0 +1,125 @@
+"""OneBot v11 duplex transport, independent of an HTTP server."""
+
+from __future__ import annotations
+
+import asyncio
+import functools
+import json
+import uuid
+
+from aiocqhttp.exceptions import ActionFailed, ApiNotAvailable
+
+
+class OneBotConnection:
+    def __init__(self, send, receive, close, on_event, validate, *, self_id='', timeout=30):
+        self.send = send
+        self.receive = receive
+        self.close_socket = close
+        self.on_event = on_event
+        self.validate = validate
+        self.self_id = self_id
+        self.timeout = timeout
+        self.pending = {}
+        self.events = asyncio.Queue(maxsize=256)
+        self.closed = False
+        self.send_lock = asyncio.Lock()
+
+    async def call_action(self, action, **params):
+        await self.validate()
+        if self.closed:
+            raise ApiNotAvailable('OneBot connection closed')
+        if len(self.pending) >= 256:
+            raise ApiNotAvailable('Too many pending OneBot actions')
+        echo = uuid.uuid4().hex
+        future = asyncio.get_running_loop().create_future()
+        self.pending[echo] = future
+        try:
+            async with asyncio.timeout(self.timeout):
+                async with self.send_lock:
+                    await self.validate()
+                    await self.send(json.dumps({'action': action, 'params': params, 'echo': echo}))
+                result = await future
+            if result.get('status') not in ('ok', 'async') or result.get('retcode') not in (0, 1):
+                raise ActionFailed(result)
+            return result.get('data')
+        finally:
+            self.pending.pop(echo, None)
+            if not future.done():
+                future.cancel()
+
+    async def _read(self):
+        while True:
+            raw = await self.receive()
+            if len(raw) > 8 * 1024 * 1024:
+                raise ValueError('OneBot frame too large')
+            data = json.loads(raw)
+            if not isinstance(data, dict):
+                raise ValueError('OneBot frame must be an object')
+            await self.validate()
+            if 'post_type' in data:
+                account = str(data.get('self_id', ''))
+                if not account or (self.self_id and account != self.self_id):
+                    raise ValueError('OneBot account mismatch')
+                self.self_id = account
+                # Never block responses behind an event that itself calls an API.
+                self.events.put_nowait(data)
+            else:
+                echo = data.get('echo')
+                future = self.pending.get(echo) if isinstance(echo, str) else None
+                if future is not None and not future.done():
+                    future.set_result(data)
+
+    async def _consume(self):
+        while True:
+            data = await self.events.get()
+            await self.validate()
+            await self.on_event(data)
+
+    async def _watch_placement(self):
+        while True:
+            await asyncio.sleep(5)
+            await self.validate()
+
+    async def run(self):
+        tasks = [asyncio.create_task(coro()) for coro in (self._read, self._consume, self._watch_placement)]
+        try:
+            done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
+            for task in done:
+                task.result()
+        finally:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await self.close()
+
+    async def close(self):
+        if self.closed:
+            return
+        self.closed = True
+        for future in self.pending.values():
+            if not future.done():
+                future.set_exception(ApiNotAvailable('OneBot connection closed'))
+        self.pending.clear()
+        await self.close_socket()
+
+
+class OneBotClient:
+    """Keep the existing converter/API call surface without CQHttp's server."""
+
+    def __init__(self):
+        self.connection = None
+
+    async def call_action(self, action, **params):
+        connection = self.connection
+        if connection is None:
+            raise ApiNotAvailable('OneBot protocol endpoint is not connected')
+        return await connection.call_action(action, **params)
+
+    def __getattr__(self, action):
+        if action.startswith('_'):
+            raise AttributeError(action)
+        return functools.partial(self.call_action, action)
+
+    async def send(self, event, message, **kwargs):
+        params = {key: event[key] for key in ('message_type', 'user_id', 'group_id', 'discuss_id') if key in event}
+        return await self.call_action('send_msg', **(params | {'message': message} | kwargs))

src/langbot/pkg/platform/botmgr.py

@@ -2357,6 +2357,9 @@ class PlatformManager:
                 execution_context=execution_context,
             )
 
+            if hasattr(adapter_inst, 'bind_runtime'):
+                adapter_inst.bind_runtime(runtime_bot)
+
             await runtime_bot.initialize()
 
             bot_key = self._runtime_bot_key(runtime_bot)

tests/e2e/live_aiocqhttp_eba_probe.py

@@ -7,6 +7,7 @@ import json
 import time
 from collections import Counter
 from pathlib import Path
+from types import SimpleNamespace
 
 from langbot.pkg.platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter
 from langbot_plugin.api.definition.abstract.platform.event_logger import AbstractEventLogger
@@ -36,9 +37,8 @@ def dump_event(event: platform_events.Event) -> dict:
 
 async def main():
     parser = argparse.ArgumentParser(description='Live OneBot v11 / aiocqhttp EBA probe for Matcha or a real endpoint.')
-    parser.add_argument('--host', default='127.0.0.1')
-    parser.add_argument('--port', type=int, default=2280)
-    parser.add_argument('--access-token', default='')
+    parser.add_argument('--ws-url', required=True, help='Protocol endpoint Universal WebSocket URL')
+    parser.add_argument('--access-token', required=True)
     parser.add_argument('--timeout', type=int, default=120)
     parser.add_argument('--target-type', choices=['private', 'group'], default=None)
     parser.add_argument('--target-id', default=None)
@@ -54,10 +54,18 @@ async def main():
     out_fp = out_path.open('a', encoding='utf-8')
 
     adapter = AiocqhttpAdapter(
-        {'host': args.host, 'port': args.port, 'access-token': args.access_token},
+        {'connection_mode': 'forward_ws', 'ws_url': args.ws_url, 'access-token': args.access_token},
         ProbeLogger(),
     )
 
+    # This standalone converter probe uses a single explicit local runtime.
+    runtime = SimpleNamespace(enable=True, workspace_uuid='probe', bot_entity=SimpleNamespace(uuid='probe'))
+
+    async def resolve_probe(_):
+        return runtime
+
+    runtime.ap = SimpleNamespace(platform_mgr=SimpleNamespace(resolve_public_bot=resolve_probe))
+    adapter.bind_runtime(runtime)
     observed: list[platform_events.Event] = []
     first_message = asyncio.Event()
 
@@ -79,7 +87,7 @@ async def main():
             return None
 
     task = asyncio.create_task(adapter.run_async())
-    print(f'Listening on ws://{args.host}:{args.port}/ws/ . Trigger events from Matcha now.')
+    print('Connecting to the protocol endpoint. Trigger events now.')
 
     api_results: dict[str, str] = {}
     try:

tests/unit_tests/platform/test_aiocqhttp_eba_adapter.py

@@ -34,7 +34,7 @@ class DummyLogger(AbstractEventLogger):
 
 
 def make_adapter() -> AiocqhttpAdapter:
-    return AiocqhttpAdapter({'host': '127.0.0.1', 'port': 2280, 'access-token': ''}, DummyLogger())
+    return AiocqhttpAdapter({'host': '127.0.0.1', 'port': 2280, 'access-token': 'test-secret'}, DummyLogger())
 
 
 def manifest() -> dict:
@@ -48,13 +48,40 @@ def manifest() -> dict:
         / 'aiocqhttp'
         / 'manifest.yaml'
     )
-    return yaml.safe_load(manifest_path.read_text())
+    return yaml.safe_load(manifest_path.read_text(encoding='utf-8'))
 
 
 def onebot_event(payload: dict) -> aiocqhttp.Event:
     return aiocqhttp.Event.from_payload(payload)
 
 
+@pytest.mark.parametrize('meta_type', ['heartbeat', 'lifecycle', 'future_transport_metadata'])
+async def test_protocol_metadata_never_reaches_business_listeners(meta_type):
+    adapter = make_adapter()
+    listener = AsyncMock()
+    adapter.register_listener(platform_events.EBAEvent, listener)
+    payload = {'post_type': 'meta_event', 'meta_event_type': meta_type, 'self_id': 42, 'time': 1700000000}
+    assert await AiocqhttpEventConverter.target2yiri(onebot_event(payload)) is None
+    await adapter._receive_event(payload)
+    listener.assert_not_awaited()
+    assert adapter.bot_account_id == '42'
+
+    # Genuine platform-specific user interactions still reach business routing.
+    await adapter._receive_event(
+        {
+            'post_type': 'notice',
+            'notice_type': 'notify',
+            'sub_type': 'poke',
+            'self_id': 42,
+            'time': 1700000001,
+            'user_id': 10001,
+            'target_id': 42,
+        }
+    )
+    listener.assert_awaited_once()
+    assert listener.await_args.args[0].type == 'platform.specific'
+
+
 def test_aiocqhttp_supported_events_match_manifest():
     assert make_adapter().get_supported_events() == manifest()['spec']['supported_events']
 
@@ -255,7 +282,7 @@ async def test_aiocqhttp_event_converter_maps_private_and_group_messages():
 
     assert isinstance(private_event, platform_events.MessageReceivedEvent)
     assert private_event.type == 'message.received'
-    assert private_event.adapter_name == 'aiocqhttp-omni'
+    assert private_event.adapter_name == 'onebot-v11'
     assert private_event.chat_type == platform_entities.ChatType.PRIVATE
     assert private_event.chat_id == 10001
     assert private_event.sender.nickname == 'Alice'

tests/unit_tests/platform/test_botmgr_tenancy.py

@@ -243,7 +243,8 @@ async def test_reload_stops_and_drops_existing_platform_runtimes():
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ('saved_adapter', 'registered_adapter'),
-    [('probe', 'probe')] + [(f'{name}-eba', f'{name}-omni') for name in sorted(OMNI_ADAPTER_NAMES)],
+    [('probe', 'probe'), ('aiocqhttp-eba', 'onebot-v11'), ('aiocqhttp-omni', 'onebot-v11')]
+    + [(f'{name}-eba', f'{name}-omni') for name in sorted(OMNI_ADAPTER_NAMES)],
 )
 async def test_cloud_startup_reuses_validated_platform_binding(saved_adapter, registered_adapter):
     class TenantUow:

tests/unit_tests/platform/test_omni_adapter_names.py

@@ -13,13 +13,17 @@ from langbot.pkg.platform.botmgr import PlatformManager
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize('name', sorted(OMNI_ADAPTER_NAMES))
-async def test_saved_adapter_ids_are_normalized_across_bot_api(name):
-    saved = {'uuid': 'bot', 'adapter': f'{name}-eba', 'adapter_config': {}}
+@pytest.mark.parametrize(
+    'saved_name,public_name',
+    [(f'{name}-eba', f'{name}-omni') for name in sorted(OMNI_ADAPTER_NAMES)]
+    + [('aiocqhttp-eba', 'onebot-v11'), ('aiocqhttp-omni', 'onebot-v11'), ('onebot-v11', 'onebot-v11')],
+)
+async def test_saved_adapter_ids_are_normalized_across_bot_api(saved_name, public_name):
+    saved = {'uuid': 'bot', 'adapter': saved_name, 'adapter_config': {}}
     component = SimpleNamespace(
-        metadata=SimpleNamespace(name=f'{name}-omni'),
+        metadata=SimpleNamespace(name=public_name),
         spec={'config': [{'type': 'webhook-url'}]},
-        to_plain_dict=lambda: {'metadata': {'name': f'{name}-omni'}},
+        to_plain_dict=lambda: {'metadata': {'name': public_name}},
     )
     result = SimpleNamespace(all=lambda: [saved], first=lambda: saved)
     application = SimpleNamespace(
@@ -30,18 +34,20 @@ async def test_saved_adapter_ids_are_normalized_across_bot_api(name):
         ),
     )
     service = BotService(application)
-    assert (await service.get_bots('workspace'))[0]['adapter'] == f'{name}-omni'
-    assert (await service.get_bot('workspace', 'bot'))['adapter'] == f'{name}-omni'
+    assert (await service.get_bots('workspace'))[0]['adapter'] == public_name
+    assert (await service.get_bot('workspace', 'bot'))['adapter'] == public_name
     assert service._adapter_declares_webhook_url(saved['adapter'])
-    assert (await service._prepare_bot_data('workspace', saved, include_uuid=True))['adapter'] == f'{name}-omni'
-    assert saved['adapter'] == f'{name}-eba'
+    assert (await service._prepare_bot_data('workspace', saved, include_uuid=True))['adapter'] == public_name
+    assert saved['adapter'] == saved_name
     manager = PlatformManager(application)
     manager.adapter_components = [component]
     assert manager.get_available_adapter_manifest_by_name(saved['adapter']) is component
-    assert manager.get_available_adapter_info_by_name(saved['adapter'])['metadata']['name'] == f'{name}-omni'
+    assert manager.get_available_adapter_info_by_name(saved['adapter'])['metadata']['name'] == public_name
 
 
-@pytest.mark.parametrize('name', ['telegram', 'custom-eba', 'my-telegram-eba', 'telegram-omni', 'websocket'])
+@pytest.mark.parametrize(
+    'name', ['aiocqhttp', 'onebot-v11', 'telegram', 'custom-eba', 'my-telegram-eba', 'telegram-omni', 'websocket']
+)
 def test_legacy_and_custom_adapter_ids_are_unchanged(name):
     assert canonical_adapter_name(name) == name
 
@@ -49,14 +55,23 @@ def test_legacy_and_custom_adapter_ids_are_unchanged(name):
 def test_discovered_manifests_expose_only_omni_names():
     root = Path(__file__).resolve().parents[3] / 'src/langbot/pkg/platform/adapters'
     for name in OMNI_ADAPTER_NAMES:
-        metadata = yaml.safe_load((root / name / 'manifest.yaml').read_text())['metadata']
+        metadata = yaml.safe_load((root / name / 'manifest.yaml').read_text(encoding='utf-8'))['metadata']
         assert metadata['name'] == f'{name}-omni'
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize('disabled_name', ['telegram-eba', 'telegram-omni'])
-async def test_old_disabled_adapter_config_still_hides_omni(disabled_name):
-    component = SimpleNamespace(metadata=SimpleNamespace(name='telegram-omni'))
+@pytest.mark.parametrize(
+    'disabled_name,public_name',
+    [
+        ('telegram-eba', 'telegram-omni'),
+        ('telegram-omni', 'telegram-omni'),
+        ('aiocqhttp-eba', 'onebot-v11'),
+        ('aiocqhttp-omni', 'onebot-v11'),
+        ('onebot-v11', 'onebot-v11'),
+    ],
+)
+async def test_old_disabled_adapter_config_still_hides_omni(disabled_name, public_name):
+    component = SimpleNamespace(metadata=SimpleNamespace(name=public_name))
     application = SimpleNamespace(
         storage_mgr=SimpleNamespace(storage_provider=SimpleNamespace(delete_dir_recursive=AsyncMock())),
         instance_config=SimpleNamespace(data={'system': {'disabled_adapters': [disabled_name]}}),

tests/unit_tests/platform/test_onebot_transport.py

@@ -0,0 +1,329 @@
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from quart import Quart
+import quart_cors
+from quart.testing.connections import WebsocketResponseError
+from aiocqhttp.exceptions import ActionFailed, ApiNotAvailable
+from websockets.asyncio.server import serve
+
+from langbot.pkg.platform.adapters.aiocqhttp.transport import OneBotConnection
+from langbot.pkg.api.http.controller.groups.webhooks import WebhookRouterGroup
+from tests.unit_tests.platform.test_aiocqhttp_eba_adapter import make_adapter
+
+
+def bind(adapter):
+    runtime = SimpleNamespace(
+        adapter=adapter, enable=True, workspace_uuid='workspace-a', bot_entity=SimpleNamespace(uuid='bot-a')
+    )
+    ap = SimpleNamespace(platform_mgr=SimpleNamespace(resolve_public_bot=AsyncMock(return_value=runtime)))
+    runtime.ap = ap
+    adapter.bind_runtime(runtime)
+    adapter.config['access-token'] = 'secret'
+    return runtime, ap
+
+
+async def test_echo_out_of_order_and_event_api_does_not_deadlock():
+    incoming, outgoing = asyncio.Queue(), asyncio.Queue()
+    handled = asyncio.Event()
+
+    async def event(data):
+        assert await connection.call_action('get_group_info', group_id=1) == {'name': 'group'}
+        handled.set()
+
+    connection = OneBotConnection(outgoing.put, incoming.get, AsyncMock(), event, AsyncMock())
+    task = asyncio.create_task(connection.run())
+    first = asyncio.create_task(connection.call_action('first'))
+    second = asyncio.create_task(connection.call_action('second'))
+    a, b = json.loads(await outgoing.get()), json.loads(await outgoing.get())
+    for request in (b, a):
+        await incoming.put(
+            json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': request['action']})
+        )
+    assert await first == 'first'
+    assert await second == 'second'
+    await incoming.put(json.dumps({'post_type': 'message', 'self_id': 42}))
+    request = json.loads(await outgoing.get())
+    await incoming.put(json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': {'name': 'group'}}))
+    await asyncio.wait_for(handled.wait(), 1)
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+    assert not connection.pending
+
+
+async def test_timeout_disconnect_and_failure():
+    outgoing = asyncio.Queue()
+    connection = OneBotConnection(outgoing.put, AsyncMock(), AsyncMock(), AsyncMock(), AsyncMock(), timeout=0.01)
+    with pytest.raises(TimeoutError):
+        await connection.call_action('timeout')
+    assert not connection.pending
+    await outgoing.get()
+    connection.timeout = 2
+    action = asyncio.create_task(connection.call_action('failure'))
+    request = json.loads(await outgoing.get())
+    connection.pending[request['echo']].set_result({'status': 'failed', 'retcode': 100})
+    with pytest.raises(ActionFailed):
+        await action
+    action = asyncio.create_task(connection.call_action('disconnect'))
+    await outgoing.get()
+    await connection.close()
+    with pytest.raises(ApiNotAvailable):
+        await action
+
+
+@pytest.mark.parametrize('payload', ['[]', 'broken', '{"post_type":"message","self_id":99}'])
+async def test_invalid_frames_and_account_mismatch(payload):
+    incoming = asyncio.Queue()
+    await incoming.put(payload)
+    connection = OneBotConnection(AsyncMock(), incoming.get, AsyncMock(), AsyncMock(), AsyncMock(), self_id='42')
+    with pytest.raises(ValueError):
+        await connection.run()
+    assert connection.closed
+
+
+async def test_main_port_reverse_ws_auth_and_placement():
+    adapter = make_adapter()
+    runtime, ap = bind(adapter)
+    app = Quart(__name__)
+    quart_cors.cors(app, allow_origin='*')
+    await WebhookRouterGroup(ap, app).initialize()
+    client = app.test_client()
+    headers = {'Authorization': 'Bearer secret', 'X-Self-ID': '42', 'X-Client-Role': 'Universal'}
+    for changes, status in [({'Authorization': 'Bearer wrong'}, 401), ({'X-Client-Role': 'Event'}, 400)]:
+        with pytest.raises(WebsocketResponseError) as error:
+            async with client.websocket('/bots/bot-a/onebot/v11/ws', headers=headers | changes):
+                await asyncio.sleep(0.01)
+        assert error.value.response.status_code == status
+    async with client.websocket('/bots/bot-a/onebot/v11/ws', headers=headers | {'X-Workspace-Id': 'attacker'}) as ws:
+        await ws.send(json.dumps({'post_type': 'meta_event', 'meta_event_type': 'heartbeat', 'self_id': 42}))
+        for _ in range(100):
+            if adapter.bot.connection:
+                break
+            await asyncio.sleep(0.001)
+        action = asyncio.create_task(adapter.bot.get_login_info())
+        request = json.loads(await ws.receive())
+        await ws.send(json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': {'user_id': 42}}))
+        assert await asyncio.wait_for(action, 1) == {'user_id': 42}
+        ap.platform_mgr.resolve_public_bot.return_value = None
+        with pytest.raises(RuntimeError, match='placement'):
+            await adapter.bot.get_login_info()
+    assert runtime.workspace_uuid == 'workspace-a'
+    await adapter.kill()
+
+
+async def test_forward_ws_round_trip_and_shutdown():
+    connected = asyncio.Event()
+
+    async def endpoint(socket):
+        assert socket.request.headers['Authorization'] == 'Bearer secret'
+        await socket.send(json.dumps({'post_type': 'meta_event', 'meta_event_type': 'heartbeat', 'self_id': 42}))
+        connected.set()
+        async for raw in socket:
+            request = json.loads(raw)
+            await socket.send(
+                json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': {'message_id': 7}})
+            )
+
+    async with serve(endpoint, '127.0.0.1', 0) as server:
+        adapter = make_adapter()
+        bind(adapter)
+        adapter.config.update(
+            connection_mode='forward_ws', ws_url=f'ws://127.0.0.1:{server.sockets[0].getsockname()[1]}'
+        )
+        task = asyncio.create_task(adapter.run_async())
+        await asyncio.wait_for(connected.wait(), 2)
+        for _ in range(100):
+            if adapter.bot.connection:
+                break
+            await asyncio.sleep(0.001)
+        assert await adapter.bot.send_private_msg(user_id=1, message='hello') == {'message_id': 7}
+        await adapter.kill()
+        await asyncio.gather(task, return_exceptions=True)
+        assert adapter.bot.connection is None
+
+
+async def test_idle_placement_revocation_closes_connection():
+    adapter = make_adapter()
+    _, ap = bind(adapter)
+    closed = asyncio.Event()
+
+    async def close():
+        closed.set()
+
+    incoming = asyncio.Queue()
+    task = asyncio.create_task(adapter.serve_connection(AsyncMock(), incoming.get, close, '42'))
+    for _ in range(100):
+        if adapter.bot.connection:
+            break
+        await asyncio.sleep(0.001)
+    with pytest.raises(RuntimeError, match='active connection'):
+        await adapter.serve_connection(AsyncMock(), incoming.get, close, '42')
+    ap.platform_mgr.resolve_public_bot.return_value = None
+    await asyncio.wait_for(closed.wait(), 7)
+    with pytest.raises(RuntimeError, match='placement'):
+        await task
+    assert adapter.bot.connection is None
+
+
+async def test_forward_reconnects_after_endpoint_disconnect():
+    count = 0
+    reconnected = asyncio.Event()
+
+    async def endpoint(socket):
+        nonlocal count
+        count += 1
+        if count == 1:
+            await socket.close()
+        else:
+            reconnected.set()
+            await socket.wait_closed()
+
+    async with serve(endpoint, '127.0.0.1', 0) as server:
+        adapter = make_adapter()
+        bind(adapter)
+        adapter.config.update(
+            connection_mode='forward_ws', ws_url=f'ws://127.0.0.1:{server.sockets[0].getsockname()[1]}'
+        )
+        task = asyncio.create_task(adapter.run_async())
+        await asyncio.wait_for(reconnected.wait(), 4)
+        await adapter.kill()
+        await asyncio.gather(task, return_exceptions=True)
+        assert count == 2
+
+
+def test_legacy_manifest_keeps_independent_adapter():
+    from langbot.pkg.discover.engine import ComponentDiscoveryEngine
+    from langbot.pkg.platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter as NewAdapter
+    from langbot.pkg.platform.sources.aiocqhttp import AiocqhttpAdapter as LegacyAdapter
+
+    engine = ComponentDiscoveryEngine(SimpleNamespace())
+    legacy = engine.load_component_manifest('pkg/platform/sources/aiocqhttp.yaml')
+    modern = engine.load_component_manifest('pkg/platform/adapters/aiocqhttp/manifest.yaml')
+    assert legacy.get_python_component_class() is LegacyAdapter
+    assert modern.get_python_component_class() is NewAdapter
+    assert modern.metadata.name == 'onebot-v11'
+    assert LegacyAdapter is not NewAdapter
+    fields = {item['name']: item for item in legacy.spec['config']}
+    assert fields['port']['default'] == 2280
+    assert fields['access-token']['required'] is False
+    assert 'connection_mode' not in fields
+
+
+def test_configuration_rejects_invalid_credentials_and_url():
+    from langbot.pkg.platform.adapters.aiocqhttp.adapter import AiocqhttpAdapter
+    from tests.unit_tests.platform.test_aiocqhttp_eba_adapter import DummyLogger
+
+    with pytest.raises(ValueError, match='token'):
+        AiocqhttpAdapter({}, DummyLogger())
+    for url in ('https://example.com', 'ws://user:[email protected]', ''):
+        with pytest.raises(ValueError, match='ws_url'):
+            AiocqhttpAdapter({'access-token': 'secret', 'connection_mode': 'forward_ws', 'ws_url': url}, DummyLogger())
+
+
+@pytest.mark.parametrize('mode', ['reverse_ws', 'forward_ws', 'legacy'])
+async def test_message_event_to_reply_roundtrip(mode):
+    """Exercise real converters and callbacks against a simulated protocol peer."""
+    from langbot.pkg.platform.sources.aiocqhttp import AiocqhttpAdapter as LegacyAdapter
+    from langbot_plugin.api.entities.builtin.platform import events, message
+    from tests.unit_tests.platform.test_aiocqhttp_eba_adapter import DummyLogger
+
+    adapter = LegacyAdapter({'access-token': 'secret'}, DummyLogger()) if mode == 'legacy' else make_adapter()
+    finished = asyncio.Event()
+    received = []
+
+    async def callback(event, source):
+        received.append(str(event.message_chain))
+        await source.send_message('person', '10001', message.MessageChain([message.Plain(text='roundtrip reply')]))
+        finished.set()
+
+    adapter.register_listener(events.FriendMessage, callback)
+
+    async def protocol_peer(send, receive):
+        await send(
+            json.dumps(
+                {
+                    'time': 1700000000,
+                    'self_id': 42,
+                    'post_type': 'message',
+                    'message_type': 'private',
+                    'sub_type': 'friend',
+                    'message_id': 9,
+                    'user_id': 10001,
+                    'message': [{'type': 'text', 'data': {'text': 'roundtrip incoming'}}],
+                    'raw_message': 'roundtrip incoming',
+                    'font': 0,
+                    'sender': {'user_id': 10001, 'nickname': 'Protocol test'},
+                }
+            )
+        )
+        while not finished.is_set():
+            request = json.loads(await asyncio.wait_for(receive(), 3))
+            action = request['action']
+            if action == 'send_private_msg':
+                assert str(request['params']['user_id']) == '10001'
+                assert 'roundtrip reply' in json.dumps(request['params']['message'])
+                data = {'message_id': 10}
+            elif action == 'get_stranger_info':
+                data = {'user_id': 10001, 'nickname': 'Protocol test'}
+            else:
+                raise AssertionError(f'Unexpected action: {action}')
+            await send(json.dumps({'echo': request['echo'], 'status': 'ok', 'retcode': 0, 'data': data}))
+            if action == 'send_private_msg':
+                await asyncio.wait_for(finished.wait(), 3)
+
+    if mode == 'forward_ws':
+        bind(adapter)
+
+        async def endpoint(ws):
+            assert ws.request.headers['Authorization'] == 'Bearer secret'
+            await protocol_peer(ws.send, ws.recv)
+
+        async with serve(endpoint, '127.0.0.1', 0) as server:
+            adapter.config.update(connection_mode=mode, ws_url=f'ws://127.0.0.1:{server.sockets[0].getsockname()[1]}')
+            task = asyncio.create_task(adapter.run_async())
+            try:
+                await asyncio.wait_for(finished.wait(), 5)
+            finally:
+                await adapter.kill()
+                await asyncio.gather(task, return_exceptions=True)
+    else:
+        if mode == 'legacy':
+            app, path = adapter.bot._server_app, '/ws/'
+        else:
+            _, ap = bind(adapter)
+            app, path = Quart(__name__), '/bots/bot-a/onebot/v11/ws'
+            await WebhookRouterGroup(ap, app).initialize()
+        headers = {'Authorization': 'Bearer secret', 'X-Self-ID': '42', 'X-Client-Role': 'Universal'}
+        async with app.test_client().websocket(path, headers=headers) as ws:
+            await protocol_peer(ws.send, ws.receive)
+        await adapter.kill()
+    assert received == ['roundtrip incoming']
+
+
+@pytest.mark.parametrize(
+    'name,path',
+    [
+        ('aiocqhttp', 'pkg/platform/sources/aiocqhttp.yaml'),
+        ('onebot-v11', 'pkg/platform/adapters/aiocqhttp/manifest.yaml'),
+        ('aiocqhttp-omni', 'pkg/platform/adapters/aiocqhttp/manifest.yaml'),
+        ('aiocqhttp-eba', 'pkg/platform/adapters/aiocqhttp/manifest.yaml'),
+    ],
+)
+async def test_public_url_only_available_for_new_onebot_adapter(name, path):
+    from langbot.pkg.discover.engine import ComponentDiscoveryEngine
+    from langbot.pkg.api.http.service.bot import BotService
+
+    ap = SimpleNamespace(
+        instance_config=SimpleNamespace(data={'api': {'webhook_prefix': 'https://bot.example.com'}}),
+        platform_mgr=SimpleNamespace(get_bot_by_uuid=AsyncMock(return_value=None)),
+    )
+    ap.discover = ComponentDiscoveryEngine(ap)
+    ap.discover.load_component_manifest(path)
+    service = BotService(ap)
+    service.get_bot = AsyncMock(return_value={'uuid': 'bot-a', 'adapter': name, 'adapter_config': {}})
+    result = await service.get_runtime_bot_info('workspace-a', 'bot-a')
+    expected = 'https://bot.example.com/bots/bot-a' if name != 'aiocqhttp' else None
+    assert result['adapter_runtime_values']['webhook_full_url'] == expected

web/src/app/home/bots/components/bot-form/BotForm.tsx

@@ -288,6 +288,10 @@ const BotForm = forwardRef<BotFormHandle, BotFormProps>(function BotForm(
     () => ({
       bot_uuid: initBotId,
       webhook_url: webhookUrl,
+      onebot_ws_url: webhookUrl
+        ? webhookUrl.replace(/^http/, 'ws').replace(/\/$/, '') +
+          '/onebot/v11/ws'
+        : '',
       extra_webhook_url: extraWebhookUrl,
       adapter_config: { title: embedTitle },
       outbound_ips: systemInfo.outbound_ips,

web/src/app/wizard/page.tsx

@@ -1823,6 +1823,10 @@ function StepBotConfig({
                   systemContext={{
                     is_wizard: true,
                     webhook_url: webhookUrl,
+                    onebot_ws_url: webhookUrl
+                      ? webhookUrl.replace(/^http/, 'ws').replace(/\/$/, '') +
+                        '/onebot/v11/ws'
+                      : '',
                     extra_webhook_url: extraWebhookUrl,
                     outbound_ips: systemInfo.outbound_ips,
                   }}