feat(btw): submit explicit work tasks through builtin commands

astrbot/api/__init__.py

@@ -4,6 +4,9 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from astrbot.core.agent.btw.types import (
+        is_work_loop_enabled as btw_work_loop_enabled,
+    )
     from astrbot.core.agent.tool import FunctionTool, ToolSet
     from astrbot.core.agent.tool_executor import BaseFunctionToolExecutor
     from astrbot.core.auth import AuthContext, Decision, Resource, Role, Subject
@@ -13,6 +16,10 @@
     from astrbot.core.utils.error_redaction import safe_error
 
 _EXPORTS = {
+    "btw_work_loop_enabled": (
+        "astrbot.core.agent.btw.types",
+        "is_work_loop_enabled",
+    ),
     "AuthContext": ("astrbot.core.auth", "AuthContext"),
     "Decision": ("astrbot.core.auth", "Decision"),
     "Resource": ("astrbot.core.auth", "Resource"),
@@ -78,6 +85,7 @@ def __getattr__(self, item: str):
     "Subject",
     "ToolSet",
     "agent",
+    "btw_work_loop_enabled",
     "llm_tool",
     "logger",
     "safe_error",

astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/en-US.json

@@ -4,6 +4,8 @@
     "desc": "AstrBot built-in session, conversation, provider, persona, plugin, and bot commands."
   },
   "commands": {
+    "work.disabled": "The BTW work loop is not enabled.",
+    "work.usage": "Usage: /work <task>",
     "help.header": "AstrBot v{version} (WebUI: {dashboard})",
     "help.empty": "No enabled built-in commands.",
     "help.tip": "Tip: use `/help --image` to render the visual help card.",

astrbot/builtin_stars/builtin_commands/.astrbot-plugin/i18n/zh-CN.json

@@ -4,6 +4,8 @@
     "desc": "AstrBot 内置的会话、对话、Provider、Persona、插件与机器人指令。"
   },
   "commands": {
+    "work.disabled": "BTW 工作循环尚未启用。",
+    "work.usage": "用法：/work <任务内容>",
     "help.header": "AstrBot v{version}（WebUI：{dashboard}）",
     "help.empty": "当前没有已启用的内置指令。",
     "help.tip": "提示：使用 `/help --image` 可生成图片版帮助。",

astrbot/builtin_stars/builtin_commands/commands/__init__.py

@@ -11,6 +11,7 @@
 from .provider import ProviderCommands
 from .session import SessionCommands
 from .variable import VariableCommands
+from .work import WorkCommands
 
 __all__ = [
     "AdminCommands",
@@ -24,4 +25,5 @@
     "ProviderCommands",
     "SessionCommands",
     "VariableCommands",
+    "WorkCommands",
 ]

astrbot/builtin_stars/builtin_commands/commands/work.py

@@ -0,0 +1,33 @@
+"""Explicit submission of free-text tasks to the BTW work loop."""
+
+from astrbot.api import btw_work_loop_enabled
+from astrbot.api.event import AstrMessageEvent
+
+from .reply import reply_i18n
+
+
+class WorkCommands:
+    """The built-in work-loop command surface."""
+
+    def __init__(self, context) -> None:
+        self.context = context
+
+    async def handle(self, event: AstrMessageEvent, task: str = "") -> None:
+        """Submit a task, reserving empty input and ``status`` for queries."""
+        stripped = (task or "").strip()
+        if not stripped or stripped.lower() == "status":
+            await reply_i18n(self.context, event, "work.usage")
+            return
+        await self.submit(event, stripped)
+
+    async def submit(self, event: AstrMessageEvent, task: str) -> None:
+        """Continue the admitted command event through the work loop."""
+        config = self.context.config.get(umo=event.unified_msg_origin)
+        if not btw_work_loop_enabled(config):
+            await reply_i18n(self.context, event, "work.disabled")
+            return
+        event.message_str = task
+        event.set_extra("should_run_command", False)
+        event.set_extra("should_run_llm", True)
+        event.set_extra("btw_force_work", True)
+        event.set_extra("btw_loop", "work")

astrbot/builtin_stars/builtin_commands/main.py

@@ -16,6 +16,7 @@
     ProviderCommands,
     SessionCommands,
     VariableCommands,
+    WorkCommands,
 )
 
 
@@ -34,6 +35,7 @@ def __init__(self, context: star.PluginContext) -> None:
         self.provider_c = ProviderCommands(self.context)
         self.session_c = SessionCommands(self.context)
         self.variable_c = VariableCommands(self.context)
+        self.work_c = WorkCommands(self.context)
 
     @filter.command("help")
     async def help(
@@ -108,6 +110,16 @@ async def conversation_reset(self, message: AstrMessageEvent) -> None:
     def task(self) -> None:
         """Manage running tasks"""
 
+    @filter.permission("session.read")
+    @filter.command("work")
+    async def work(
+        self,
+        event: AstrMessageEvent,
+        task: GreedyStr = GreedyStr(""),
+    ) -> None:
+        """Submit a task to the BTW work loop"""
+        await self.work_c.handle(event, task)
+
     @filter.permission("session.manage")
     @task.command("stop")
     async def task_stop(self, message: AstrMessageEvent) -> None:

docs/en/dev/architecture.md

@@ -184,6 +184,8 @@ Core diagnostics retain only stable error codes, Unicode code-point spans, param
 
 ## Agents, Tools, and Skills
 
+`/work <task>` deliberately accepts a greedy remainder instead of a verb subcommand: it is an explicit entry into the work loop. It still uses the native command schema, `session.read` authorization, and `builtin_commands:work` identity. The built-in handler reads enablement through the lazy `astrbot.api.btw_work_loop_enabled` helper and continues the same event through `ProcessStage` and `ConversationLoop`.
+
 The Agent runtime is under `astrbot/core/agent/`, with main-request assembly in `astrbot/core/astr_main_agent.py`. Provider abstractions live in `astrbot/core/provider/`; concrete OpenAI, Anthropic, Gemini, and similar sources live in `provider/sources/` and are lazily registered through `provider_modules.py`. Dify, Coze, DashScope, and DeerFlow are external Agent Runners under `astrbot/core/agent/runners/`, not ordinary model providers.
 
 Tools can come from the core, plugins, or MCP. MCP supports stdio and Streamable HTTP only. Remote HTTP connections reject localhost, private, link-local, and reserved addresses by default; a trusted configuration must explicitly set `allow_private_network` to opt in.

docs/en/use/command.md

@@ -81,6 +81,7 @@ The user ID from `/session info` can be granted current-session `session_admin`
 
 ### Running Tasks
 
+- `/work <task>`: Submit the remaining text to the BTW work loop without a `/chat` prefix or automatic classification. Requires `session.read`, with `btw.enabled` and `btw.work_loop.enabled` enabled on the current profile. The command identity is `builtin_commands:work`. Empty input and a lone `status` show usage; `status refactor` is ordinary task text. Command quoting rules still apply.
 - `/task stop`: Stop running Agent or third-party Agent Runner tasks in the current session without deleting history.
 
 ### Providers and Models

docs/zh/dev/architecture.md

@@ -184,6 +184,8 @@ Mixin 通过带类型的 `store_session(self)` 助手获取会话，不直接持
 
 ## Agent、工具与 Skills
 
+`/work <任务内容>` 有意使用贪婪文本参数，而非动词子命令，作为工作循环的显式入口。它仍使用原生指令 schema、`session.read` 授权和 `builtin_commands:work` 标识。内置 handler 通过延迟加载的 `astrbot.api.btw_work_loop_enabled` 查询启用状态，并将同一事件继续交给 `ProcessStage` 与 `ConversationLoop`。
+
 核心 Agent 运行时位于 `astrbot/core/agent/`，主 Agent 的请求组装位于 `astrbot/core/astr_main_agent.py`。Provider 抽象位于 `astrbot/core/provider/`；OpenAI、Anthropic、Gemini 等具体实现位于 `provider/sources/`，并通过 `provider_modules.py` 延迟注册。Dify、Coze、DashScope 和 DeerFlow 属于 `astrbot/core/agent/runners/` 下的外部 Agent Runner，不是普通模型 Provider。
 
 工具来源包括内置工具、插件工具和 MCP 工具。MCP 仅支持 stdio 与 Streamable HTTP；远程 HTTP 默认拒绝 localhost、私网、链路本地和保留地址，只有在可信配置中显式设置 `allow_private_network` 才会放开。

docs/zh/use/command.md

@@ -81,6 +81,7 @@ Orbit 不执行变量、命令、算术或波浪号展开，也不执行 glob、
 
 ### 运行任务
 
+- `/work <任务内容>`：将后面的文本显式提交给 BTW 工作循环，不需要 `/chat` 前缀或自动分类。要求 `session.read`，并在当前配置中启用 `btw.enabled` 和 `btw.work_loop.enabled`。指令标识为 `builtin_commands:work`。空参数和单独的 `status` 显示用法，`status 重构` 则作为普通任务文本处理；仍遵循指令引号规则。
 - `/task stop`：停止当前会话中正在运行的 Agent 或第三方 Agent Runner 任务，不删除历史。
 
 ### Provider 与模型

tests/unit/test_builtin_command_extensions.py

@@ -15,6 +15,7 @@
 from astrbot.builtin_stars.builtin_commands.commands.persona import PersonaCommands
 from astrbot.builtin_stars.builtin_commands.commands.plugin import PluginCommands
 from astrbot.builtin_stars.builtin_commands.commands.provider import ProviderCommands
+from astrbot.builtin_stars.builtin_commands.commands.work import WorkCommands
 from astrbot.builtin_stars.builtin_commands.main import Main
 from astrbot.core.command import (
     CommandEngine,
@@ -121,6 +122,115 @@ def _plain_text(result) -> str:
     return result.chain[0].text
 
 
+@pytest.mark.parametrize(
+    ("text", "expected"),
+    [
+        ("work", ""),
+        ("work status", "status"),
+        ("work STATUS", "STATUS"),
+        ("work refactor this module", "refactor this module"),
+        ("work status refactor", "status refactor"),
+        ('work "inspect the file"', "inspect the file"),
+    ],
+)
+def test_work_command_binds_the_complete_task(text, expected):
+    from astrbot.builtin_stars.builtin_commands import main as builtin_commands_main
+
+    declarations = collect_plugin_module_declarations(builtin_commands_main)
+    handlers = materialize_handler_declarations(list(declarations.handlers))
+    engine = CommandEngine(build_command_catalog(handlers))
+    result = engine.resolve(text)
+    assert result.resolution.command_path == ("work",)
+    assert dict(engine.bind(result.resolution.entries[0], result).values) == {
+        "task": expected
+    }
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "config",
+    [
+        None,
+        {},
+        {"btw": "yes"},
+        {"btw": {"enabled": False, "work_loop": {"enabled": True}}},
+        {"btw": {"enabled": True, "work_loop": {"enabled": False}}},
+    ],
+)
+async def test_work_submit_requires_both_loop_switches(config):
+    command = WorkCommands(
+        SimpleNamespace(config=SimpleNamespace(get=lambda **_: config), i18n=FakeI18n())
+    )
+    event = DummyEvent(message_str="work inspect the file")
+    await command.handle(event, "inspect the file")
+    assert _plain_text(event.result) == "The BTW work loop is not enabled."
+    assert event.is_stopped()
+    assert event.get_extra("btw_force_work") is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("task", ["", "status", " STATUS "])
+async def test_work_empty_and_status_remainders_show_usage(task):
+    command = WorkCommands(SimpleNamespace(i18n=FakeI18n()))
+    event = DummyEvent(message_str="work " + task)
+    await command.handle(event, task)
+    assert _plain_text(event.result) == "Usage: /work <task>"
+    assert event.is_stopped()
+    assert event.get_extra("btw_force_work") is None
+
+
+@pytest.mark.asyncio
+async def test_work_submission_continues_through_process_into_work_loop():
+    from astrbot.core.agent.conversation_loop import ConversationLoop
+    from astrbot.core.pipeline.process_stage.stage import ProcessStage
+
+    profile = {
+        "provider_settings": {"enable": True},
+        "btw": {"enabled": True, "work_loop": {"enabled": True}},
+    }
+    command = WorkCommands(
+        SimpleNamespace(
+            config=SimpleNamespace(get=lambda **_: profile), i18n=FakeI18n()
+        )
+    )
+
+    class Agent:
+        async def initialize(self, ctx):
+            pass
+
+        async def process(self, event):
+            received.append((event.message_str, event.get_extra("btw_loop")))
+            yield
+
+    class Handler:
+        async def process(self, event):
+            await command.handle(event, "status inspect the file")
+            yield
+
+    received = []
+    agent = Agent()
+    loop = ConversationLoop(agent)
+    await loop.initialize(SimpleNamespace(astrbot_config=profile))
+    stage = ProcessStage()
+    stage.ctx = SimpleNamespace(astrbot_config=profile)
+    stage.agent_sub_stage = agent
+    stage.conversation_loop = loop
+    stage.star_request_sub_stage = Handler()
+    event = DummyEvent(message_str="work status inspect the file")
+    event._has_send_oper = False
+    event.get_result = lambda: event.result
+    event.set_extra("activated_handlers", [object()])
+
+    _ = [part async for part in stage.process(event)]
+
+    assert received == [("status inspect the file", "work")]
+    assert event.get_extra("should_run_command") is False
+    assert event.get_extra("btw_force_work") is True
+    assert not event.is_stopped()
+    assert event.result is None
+    await loop.close()
+
+
 def test_all_builtin_extension_commands_use_native_command_schemas():
     expected_handlers = {
         "admin_list",
@@ -162,6 +272,7 @@ def test_all_builtin_extension_commands_use_native_command_schemas():
         "provider_set_stt",
         "provider_set_tts",
         "task_stop",
+        "work",
         "variable_set",
         "variable_unset",
         "flow_enable",
@@ -983,6 +1094,7 @@ def test_non_public_builtin_commands_declare_the_planned_actions():
         "bot_disable": "session.manage",
         "bot_leave": "session.manage",
         "task_stop": "session.manage",
+        "work": "session.read",
         "conversation_create": "session.manage",
         "conversation_stats": "session.read",
         "conversation_history": "session.read",

tests/unit/test_core_import_smoke.py

@@ -4,6 +4,30 @@
 from pathlib import Path
 
 
+def test_btw_sdk_enable_check_does_not_construct_runtime(tmp_path: Path) -> None:
+    root = tmp_path / "runtime-root"
+    environment = {**os.environ, "ASTRBOT_ROOT": str(root)}
+    code = """
+import os
+import pathlib
+import sys
+from astrbot.api import btw_work_loop_enabled
+assert btw_work_loop_enabled({'btw': {'enabled': True, 'work_loop': {'enabled': True}}})
+assert not btw_work_loop_enabled(None)
+assert 'astrbot.core.agent.btw.work_loop' not in sys.modules
+assert 'astrbot.core.pipeline.scheduler' not in sys.modules
+assert not pathlib.Path(os.environ['ASTRBOT_ROOT']).exists()
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        env=environment,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+
 def test_importing_core_does_not_create_runtime_services(tmp_path: Path) -> None:
     """The package boundary must stay inert in a fresh interpreter."""
     root = tmp_path / "runtime-root"