feat(btw): add opt-in conversation and work entry

astrbot/core/agent/conversation_loop.py

@@ -0,0 +1,34 @@
+"""Opt-in conversation entry over the existing Agent request executor."""
+
+from collections.abc import AsyncGenerator
+from typing import TYPE_CHECKING
+
+from astrbot.core.platform.astr_message_event import AstrMessageEvent
+
+if TYPE_CHECKING:
+    from astrbot.core.pipeline.context import PipelineContext
+    from astrbot.core.pipeline.process_stage.method.agent_request import (
+        AgentRequestSubStage,
+    )
+
+
+class ConversationLoop:
+    """Own conversation admission without choosing an automatic classifier."""
+
+    def __init__(self, agent_request: AgentRequestSubStage) -> None:
+        self.agent_request = agent_request
+        self._btw_enabled = False
+
+    async def initialize(self, ctx: PipelineContext) -> None:
+        """Initialize the shared Agent executor for this profile."""
+        self.astrbot_config = ctx.astrbot_config
+        btw = self.astrbot_config.get("btw", {})
+        self._btw_enabled = isinstance(btw, dict) and bool(btw.get("enabled", False))
+        await self.agent_request.initialize(ctx)
+
+    async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
+        """Process one admitted conversation using the current Agent path."""
+        if self._btw_enabled:
+            event.set_extra("btw_loop", "conversation")
+        async for response in self.agent_request.process(event):
+            yield response

astrbot/core/config/default.py

@@ -188,6 +188,7 @@
         ),
         "agents": [],
     },
+    "btw": {"enabled": False},
     "provider_stt_settings": {
         "enable": False,
         "provider_id": "",
@@ -4685,6 +4686,18 @@
     },
 }
 
+CONFIG_METADATA_3["plugin_group"]["metadata"]["btw"] = {
+    "description": "BTW 双循环",
+    "type": "object",
+    "items": {
+        "btw.enabled": {
+            "description": "启用 BTW 双循环",
+            "type": "bool",
+            "hint": "实验功能，默认关闭。开启后，普通 AI 请求通过对话循环进入现有 Agent。",
+        },
+    },
+}
+
 CONFIG_METADATA_3_SYSTEM = {
     "system_group": {
         "name": "系统配置",

astrbot/core/pipeline/process_stage/stage.py

@@ -1,5 +1,6 @@
 from collections.abc import AsyncGenerator
 
+from astrbot.core.agent.conversation_loop import ConversationLoop
 from astrbot.core.agent.llm_types import ProviderRequest
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
 from astrbot.core.star.star_handler import StarHandlerMetadata
@@ -15,9 +16,14 @@ async def initialize(self, ctx: PipelineContext) -> None:
         self.ctx = ctx
         self.config = ctx.astrbot_config
 
-        # initialize agent sub stage
         self.agent_sub_stage = AgentRequestSubStage()
-        await self.agent_sub_stage.initialize(ctx)
+        btw = self.config.get("btw", {})
+        if isinstance(btw, dict) and btw.get("enabled", False):
+            self.conversation_loop = ConversationLoop(self.agent_sub_stage)
+            await self.conversation_loop.initialize(ctx)
+        else:
+            self.conversation_loop = None
+            await self.agent_sub_stage.initialize(ctx)
 
         # initialize star request sub stage
         self.star_request_sub_stage = StarRequestSubStage()
@@ -64,5 +70,11 @@ async def process(
             if (
                 event.get_result() and not event.is_stopped()
             ) or not event.get_result():
-                async for _ in self.agent_sub_stage.process(event):
+                async for _ in self._dispatch_agent(event):
                     yield
+
+    def _dispatch_agent(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
+        """Use the conversation entry only when the profile enables BTW."""
+        if self.conversation_loop is not None:
+            return self.conversation_loop.process(event)
+        return self.agent_sub_stage.process(event)

dashboard/src/i18n/locales/en-US/features/config-metadata.json

@@ -1174,6 +1174,15 @@
         "description": "Available Plugins",
         "hint": "All non-disabled plugins are enabled by default. If a plugin is disabled on the plugins page, selections here will not take effect."
       }
+    },
+    "btw": {
+      "description": "BTW dual loops",
+      "btw": {
+        "enabled": {
+          "description": "Enable BTW dual loops",
+          "hint": "Experimental and disabled by default. Ordinary admitted AI requests use the conversation entry over the existing Agent."
+        }
+      }
     }
   },
   "ext_group": {

dashboard/src/i18n/locales/zh-CN/features/config-metadata.json

@@ -1168,6 +1168,15 @@
         "description": "可用插件",
         "hint": "默认启用全部未被禁用的插件。若插件在插件页面被禁用,则此处的选择不会生效。"
       }
+    },
+    "btw": {
+      "description": "BTW 双循环",
+      "btw": {
+        "enabled": {
+          "description": "启用 BTW 双循环",
+          "hint": "实验功能，默认关闭。普通且已通过准入的 AI 请求经对话入口使用现有 Agent。"
+        }
+      }
     }
   },
   "ext_group": {

docs/en/dev/astrbot-config.md

@@ -199,6 +199,12 @@ Local mode operates directly on the AstrBot host and belongs only in a trusted e
 
 Alkaid [Long-term Memory](../use/long-term-memory) currently has no enable/disable configuration. Do not treat `provider_ltm_settings` as its switch. For recent group-message injection, see [Group Chat Context Awareness](../use/group-chat-context).
 
+## BTW conversation entry
+
+`btw.enabled` defaults to `false`. Enabling it sends ordinary admitted AI requests through the conversation loop and the existing Agent executor. It does not bypass message admission, session AI switches, or plugin request handling. With BTW disabled, the pipeline directly uses the current Agent request path and retains its capabilities.
+
+Automatic classifier candidates are evaluated separately. Enabling this entry does not select an automatic routing strategy.
+
 ## WebUI and authentication
 
 Important `dashboard` defaults:

docs/zh/dev/astrbot-config.md

@@ -201,6 +201,12 @@ API Key 属于敏感配置。不要把真实 `cmd_config.json`、截图、日志
 
 Alkaid [长期记忆](../use/long-term-memory) 当前没有对应的启停配置；不要把 `provider_ltm_settings` 当作长期记忆开关。群聊近期消息注入见 [群聊上下文感知](../use/group-chat-context)。
 
+## BTW 对话入口
+
+`btw.enabled` 默认为 `false`。开启后，普通且已通过准入的 AI 请求经对话循环进入现有 Agent 执行器，不绕过消息准入、会话 AI 开关或插件请求处理。关闭 BTW 时，流水线直接使用当前 Agent 请求路径，并保留其能力。
+
+自动分类器候选将分别评估。开启此入口不会选定自动路由方案。
+
 ## WebUI 与认证
 
 `dashboard` 的关键默认值：

tests/unit/test_config.py

@@ -81,6 +81,10 @@ def test_default_config_omits_group_active_reply():
     assert "active_reply" not in DEFAULT_CONFIG["provider_ltm_settings"]
 
 
+def test_btw_conversation_entry_defaults_off():
+    assert DEFAULT_CONFIG["btw"]["enabled"] is False
+
+
 def test_platform_templates_prioritize_current_adapters_without_public_defaults():
     """Platform templates keep their intended order and loopback listeners."""
     templates = CONFIG_METADATA_2["platform_group"]["metadata"]["platform"][

tests/unit/test_config_metadata_i18n.py

@@ -134,6 +134,20 @@ def test_config_metadata_locale_trees_match() -> None:
     assert sorted(en_keys - zh_keys) == []
 
 
+def test_btw_controls_survive_dashboard_metadata_conversion() -> None:
+    converted = ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3)
+    section = converted["plugin_group"]["metadata"]["btw"]
+    assert section["description"] == "plugin_group.btw.description"
+    enabled = section["items"]["btw.enabled"]
+    assert enabled["type"] == "bool"
+    assert enabled["description"] == "plugin_group.btw.btw.enabled.description"
+    assert enabled["hint"] == "plugin_group.btw.btw.enabled.hint"
+    for locale in LOCALES:
+        catalog = _load_locale(locale)
+        assert catalog[enabled["description"]]
+        assert catalog[enabled["hint"]]
+
+
 def test_config_metadata_docs_paths_are_relative_and_preserved() -> None:
     converted = ConfigMetadataI18n.convert_to_i18n_keys(CONFIG_METADATA_3)
     ai_sections = converted["ai_group"]["metadata"]

tests/unit/test_conversation_loop.py

@@ -0,0 +1,42 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from astrbot.core.agent.conversation_loop import ConversationLoop
+
+
+class FakeAgentRequest:
+    def __init__(self) -> None:
+        self.initialize = AsyncMock()
+        self.process_calls = []
+
+    async def process(self, event):
+        self.process_calls.append(event)
+        yield "first"
+        yield "second"
+
+
+class FakeEvent:
+    def __init__(self) -> None:
+        self.extras = {}
+
+    def set_extra(self, key, value) -> None:
+        self.extras[key] = value
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("btw", [{"enabled": True}, {"enabled": False}, {}, None])
+async def test_conversation_entry_preserves_agent_execution_and_disabled_metadata(btw):
+    executor = FakeAgentRequest()
+    loop = ConversationLoop(executor)
+    ctx = SimpleNamespace(astrbot_config={"btw": btw})
+    await loop.initialize(ctx)
+    event = FakeEvent()
+
+    assert [item async for item in loop.process(event)] == ["first", "second"]
+    executor.initialize.assert_awaited_once_with(ctx)
+    assert executor.process_calls == [event]
+    assert event.extras == (
+        {"btw_loop": "conversation"} if btw and btw["enabled"] else {}
+    )

tests/unit/test_process_stage.py

@@ -2,6 +2,7 @@
 import sys
 import types
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -149,9 +150,30 @@ def _stage(
     )
     stage.star_request_sub_stage = FakeSubStage(star_responses or [])
     stage.agent_sub_stage = FakeSubStage(agent_responses or [])
+    stage.conversation_loop = None
     return stage
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("enabled", [False, True])
+async def test_process_stage_initializes_only_one_agent_with_opt_in_conversation(
+    monkeypatch, enabled
+):
+    executor = FakeSubStage([])
+    executor.initialize = AsyncMock()
+    star = SimpleNamespace(initialize=AsyncMock())
+    monkeypatch.setattr(process_stage_module, "AgentRequestSubStage", lambda: executor)
+    monkeypatch.setattr(process_stage_module, "StarRequestSubStage", lambda: star)
+    stage = process_stage_module.ProcessStage()
+    ctx = SimpleNamespace(astrbot_config={"btw": {"enabled": enabled}})
+
+    await stage.initialize(ctx)
+
+    executor.initialize.assert_awaited_once_with(ctx)
+    assert stage.agent_sub_stage is executor
+    assert (stage.conversation_loop is not None) is enabled
+
+
 @pytest.mark.asyncio
 async def test_process_stage_plugin_provider_request_routes_to_agent_and_sets_extra():
     stage = _stage(