feat(btw): query work status within the owning profile and session

astrbot/api/__init__.py

@@ -4,6 +4,9 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from astrbot.core.agent.btw.runtime_registry import (
+        latest_status as btw_work_latest_status,
+    )
     from astrbot.core.agent.btw.types import (
         is_work_loop_enabled as btw_work_loop_enabled,
     )
@@ -16,6 +19,10 @@
     from astrbot.core.utils.error_redaction import safe_error
 
 _EXPORTS = {
+    "btw_work_latest_status": (
+        "astrbot.core.agent.btw.runtime_registry",
+        "latest_status",
+    ),
     "btw_work_loop_enabled": (
         "astrbot.core.agent.btw.types",
         "is_work_loop_enabled",
@@ -86,6 +93,7 @@ def __getattr__(self, item: str):
     "ToolSet",
     "agent",
     "btw_work_loop_enabled",
+    "btw_work_latest_status",
     "llm_tool",
     "logger",
     "safe_error",

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

@@ -5,7 +5,12 @@
   },
   "commands": {
     "work.disabled": "The BTW work loop is not enabled.",
-    "work.usage": "Usage: /work <task>",
+    "work.status.none": "No BTW work task has run in this session.",
+    "work.status.pending": "Queued: {task}",
+    "work.status.running": "Running: {task}",
+    "work.status.completed": "Completed: {task}",
+    "work.status.failed": "Failed: {task}",
+    "work.status.cancelled": "Cancelled: {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

@@ -5,7 +5,12 @@
   },
   "commands": {
     "work.disabled": "BTW 工作循环尚未启用。",
-    "work.usage": "用法：/work <任务内容>",
+    "work.status.none": "本会话还没有 BTW 工作任务。",
+    "work.status.pending": "排队中：{task}",
+    "work.status.running": "执行中：{task}",
+    "work.status.completed": "已完成：{task}",
+    "work.status.failed": "已失败：{task}",
+    "work.status.cancelled": "已取消：{task}",
     "help.header": "AstrBot v{version}（WebUI：{dashboard}）",
     "help.empty": "当前没有已启用的内置指令。",
     "help.tip": "提示：使用 `/help --image` 可生成图片版帮助。",

astrbot/builtin_stars/builtin_commands/commands/work.py

@@ -1,6 +1,6 @@
 """Explicit submission of free-text tasks to the BTW work loop."""
 
-from astrbot.api import btw_work_loop_enabled
+from astrbot.api import btw_work_latest_status, btw_work_loop_enabled
 from astrbot.api.event import AstrMessageEvent
 
 from .reply import reply_i18n
@@ -13,13 +13,23 @@ 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."""
+        """Query status for empty input or ``status``, otherwise submit a task."""
         stripped = (task or "").strip()
         if not stripped or stripped.lower() == "status":
-            await reply_i18n(self.context, event, "work.usage")
+            await self.status(event)
             return
         await self.submit(event, stripped)
 
+    async def status(self, event: AstrMessageEvent) -> None:
+        """Show the latest task for the command's profile and message origin."""
+        config_id = getattr(getattr(event, "resource", None), "config_id", "") or ""
+        latest = await btw_work_latest_status(config_id, event.unified_msg_origin)
+        if latest is None:
+            await reply_i18n(self.context, event, "work.status.none")
+            return
+        request, status = latest
+        await reply_i18n(self.context, event, f"work.status.{status}", task=request)
+
     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)

astrbot/builtin_stars/builtin_commands/main.py

@@ -117,7 +117,7 @@ async def work(
         event: AstrMessageEvent,
         task: GreedyStr = GreedyStr(""),
     ) -> None:
-        """Submit a task to the BTW work loop"""
+        """Submit a BTW work task, or show the latest status"""
         await self.work_c.handle(event, task)
 
     @filter.permission("session.manage")

astrbot/core/agent/btw/runtime_registry.py

@@ -0,0 +1,38 @@
+"""Expose each pipeline's most recent work status to built-in commands."""
+
+from .types import WorkSessionStatus
+from .work_sessions import WorkSessionManager
+
+_managers: dict[str, WorkSessionManager] = {}
+
+
+def register(config_id: str, manager: WorkSessionManager) -> None:
+    """Bind a successfully initialized pipeline's work-session manager."""
+    _managers[config_id] = manager
+
+
+def unregister(config_id: str, manager: WorkSessionManager) -> None:
+    """Remove only the closing pipeline's registration."""
+    if _managers.get(config_id) is manager:
+        _managers.pop(config_id)
+
+
+async def latest_status(
+    config_id: str, origin: str
+) -> tuple[str, WorkSessionStatus] | None:
+    """Read the newest work task within the specified profile and origin.
+
+    Args:
+        config_id: The profile that owns the admitted command event.
+        origin: The event's unified message origin.
+
+    Returns:
+        The task text and status, or ``None`` when no retained task exists.
+    """
+    manager = _managers.get(config_id)
+    if manager is None:
+        return None
+    session = await manager.get_for_origin(origin)
+    if session is None:
+        return None
+    return session.request, session.status

astrbot/core/pipeline/process_stage/stage.py

@@ -1,6 +1,7 @@
 import asyncio
 from collections.abc import AsyncGenerator, Awaitable, Callable
 
+from astrbot.core.agent.btw import runtime_registry
 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
@@ -29,6 +30,10 @@ async def initialize(self, ctx: PipelineContext) -> None:
         # initialize star request sub stage
         self.star_request_sub_stage = StarRequestSubStage()
         await self.star_request_sub_stage.initialize(ctx)
+        if self.conversation_loop is not None:
+            runtime_registry.register(
+                ctx.astrbot_config_id, self.conversation_loop.work_sessions
+            )
 
     def configure_detached_work(
         self,
@@ -48,7 +53,13 @@ def configure_detached_work(
     async def close(self) -> None:
         """Reclaim work before this profile's scheduler is replaced."""
         if self.conversation_loop is not None:
-            await self.conversation_loop.close()
+            try:
+                await self.conversation_loop.close()
+            finally:
+                runtime_registry.unregister(
+                    self.ctx.astrbot_config_id,
+                    self.conversation_loop.work_sessions,
+                )
 
     async def process(
         self,

docs/en/dev/architecture.md

@@ -186,6 +186,8 @@ Core diagnostics retain only stable error codes, Unicode code-point spans, param
 
 `/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`.
 
+Empty `/work` and `/work status` query the latest task through `astrbot.api.btw_work_latest_status`. `ProcessStage` registers its session manager by configuration ID after initialization and removes only that same registration when closing. Queries are scoped to both the event's profile and UMO; there is no fallback to a different profile or persistence across reloads.
+
 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,7 +81,8 @@ 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.
+- `/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`. Command quoting rules still apply.
+- `/work` or `/work status`: Show the latest task's text and status for the current profile and session: queued, running, completed, failed, or cancelled. `status` queries only when it is the entire remainder, ignoring case; `/work status refactor` submits a task. Requires `session.read`. State is kept in memory, cleared on restart or profile reload/removal, and terminal tasks expire after `btw.work_session.max_age_seconds` (default 3600).
 - `/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

@@ -186,6 +186,8 @@ Mixin 通过带类型的 `store_session(self)` 助手获取会话，不直接持
 
 `/work <任务内容>` 有意使用贪婪文本参数，而非动词子命令，作为工作循环的显式入口。它仍使用原生指令 schema、`session.read` 授权和 `builtin_commands:work` 标识。内置 handler 通过延迟加载的 `astrbot.api.btw_work_loop_enabled` 查询启用状态，并将同一事件继续交给 `ProcessStage` 与 `ConversationLoop`。
 
+空参数的 `/work` 与 `/work status` 通过 `astrbot.api.btw_work_latest_status` 查询最新任务。`ProcessStage` 完成初始化后按配置 ID 注册会话管理器，关闭时仅移除属于自身的注册。查询同时限定事件的配置和 UMO，不回退到其他配置，也不跨重载持久化。
+
 核心 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,7 +81,8 @@ Orbit 不执行变量、命令、算术或波浪号展开，也不执行 glob、
 
 ### 运行任务
 
-- `/work <任务内容>`：将后面的文本显式提交给 BTW 工作循环，不需要 `/chat` 前缀或自动分类。要求 `session.read`，并在当前配置中启用 `btw.enabled` 和 `btw.work_loop.enabled`。指令标识为 `builtin_commands:work`。空参数和单独的 `status` 显示用法，`status 重构` 则作为普通任务文本处理；仍遵循指令引号规则。
+- `/work <任务内容>`：将后面的文本显式提交给 BTW 工作循环，不需要 `/chat` 前缀或自动分类。要求 `session.read`，并在当前配置中启用 `btw.enabled` 和 `btw.work_loop.enabled`。指令标识为 `builtin_commands:work`；仍遵循指令引号规则。
+- `/work` 或 `/work status`：查看当前配置与会话中最新任务的内容及状态：排队中、执行中、已完成、已失败或已取消。仅当参数全部为 `status` 时查询，忽略大小写；`/work status 重构` 会提交任务。要求 `session.read`。状态保存在内存中，重启或配置重载、移除后清空；终态任务按 `btw.work_session.max_age_seconds` 过期，默认 3600 秒。
 - `/task stop`：停止当前会话中正在运行的 Agent 或第三方 Agent Runner 任务，不删除历史。
 
 ### Provider 与模型

tests/unit/test_btw_status.py

@@ -0,0 +1,153 @@
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from astrbot.builtin_stars.builtin_commands.commands.work import WorkCommands
+from astrbot.core.agent.btw import runtime_registry
+from astrbot.core.agent.btw.types import WorkSessionStatus
+from astrbot.core.agent.btw.work_sessions import WorkSessionManager
+from astrbot.core.pipeline.process_stage import stage as process_stage
+from tests.unit.builtin_command_fakes import FakeI18n
+from tests.unit.test_builtin_command_extensions import DummyEvent
+
+
+@pytest.fixture(autouse=True)
+def isolated_registry(monkeypatch):
+    monkeypatch.setattr(runtime_registry, "_managers", {})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("locale", ["en-US", "zh-CN"])
+@pytest.mark.parametrize("status", list(WorkSessionStatus))
+async def test_work_status_localizes_every_terminal_and_active_state(locale, status):
+    manager = WorkSessionManager()
+    event = DummyEvent(message_str="work status")
+    event.resource = SimpleNamespace(config_id="profile-a")
+    event.set_extra("locale", locale)
+    session = await manager.create(event.unified_msg_origin, "inspect the workspace")
+    await manager.update_status(session.id, status)
+    runtime_registry.register("profile-a", manager)
+    context = SimpleNamespace(i18n=FakeI18n())
+
+    await WorkCommands(context).handle(event, "STATUS")
+
+    expected = await context.i18n.t(
+        event, f"work.status.{status}", task="inspect the workspace"
+    )
+    assert event.result.get_plain_text() == expected
+    assert "inspect the workspace" in expected
+    assert "work.status." not in expected
+    assert event.is_stopped()
+    assert event.get_extra("btw_force_work") is None
+
+
+@pytest.mark.asyncio
+async def test_latest_status_stays_with_its_profile_origin_and_latest_task():
+    first = WorkSessionManager(max_age_seconds=60)
+    second = WorkSessionManager()
+    runtime_registry.register("profile-a", first)
+    runtime_registry.register("profile-b", second)
+    older = await first.create("same-origin", "first task")
+    await first.update_status(older.id, WorkSessionStatus.RUNNING)
+    latest = await first.create("same-origin", "newest task")
+    await second.create("same-origin", "other profile")
+
+    assert await runtime_registry.latest_status("profile-a", "same-origin") == (
+        "newest task",
+        WorkSessionStatus.PENDING,
+    )
+    assert await runtime_registry.latest_status("profile-b", "same-origin") == (
+        "other profile",
+        WorkSessionStatus.PENDING,
+    )
+    assert await runtime_registry.latest_status("profile-a", "other-origin") is None
+    assert (
+        await runtime_registry.latest_status("unknown-profile", "same-origin") is None
+    )
+    await first.update_status(latest.id, WorkSessionStatus.COMPLETED)
+    latest.updated_at = datetime.now(UTC) - timedelta(seconds=61)
+    assert await runtime_registry.latest_status("profile-a", "same-origin") is None
+    assert await first.get_by_id(older.id) is older
+
+
+@pytest.mark.asyncio
+async def test_status_command_uses_resource_config_without_profile_fallback():
+    manager = WorkSessionManager()
+    event = DummyEvent(message_str="work")
+    event.resource = SimpleNamespace(config_id="profile-b")
+    await manager.create(event.unified_msg_origin, "profile-a task")
+    runtime_registry.register("profile-a", manager)
+
+    await WorkCommands(SimpleNamespace(i18n=FakeI18n())).handle(event)
+
+    assert event.result.get_plain_text() == "No BTW work task has run in this session."
+
+
+@pytest.mark.asyncio
+async def test_process_registration_replacement_and_close_are_identity_scoped(
+    monkeypatch,
+):
+    monkeypatch.setattr(process_stage.AgentRequestSubStage, "initialize", AsyncMock())
+    monkeypatch.setattr(process_stage.StarRequestSubStage, "initialize", AsyncMock())
+    context = SimpleNamespace(
+        astrbot_config_id="profile-a",
+        astrbot_config={"btw": {"enabled": True, "work_loop": {"enabled": True}}},
+    )
+    old = process_stage.ProcessStage()
+    current = process_stage.ProcessStage()
+    await old.initialize(context)
+    await old.conversation_loop.work_sessions.create("origin", "old task")
+    assert (await runtime_registry.latest_status("profile-a", "origin"))[
+        0
+    ] == "old task"
+
+    await current.initialize(context)
+    await current.conversation_loop.work_sessions.create("origin", "new task")
+    await old.close()
+    assert (await runtime_registry.latest_status("profile-a", "origin"))[
+        0
+    ] == "new task"
+    await current.close()
+    await current.close()
+    assert await runtime_registry.latest_status("profile-a", "origin") is None
+
+
+@pytest.mark.asyncio
+async def test_failed_stage_initialization_does_not_publish_a_manager(monkeypatch):
+    monkeypatch.setattr(process_stage.AgentRequestSubStage, "initialize", AsyncMock())
+    monkeypatch.setattr(
+        process_stage.StarRequestSubStage,
+        "initialize",
+        AsyncMock(side_effect=RuntimeError("initialization failed")),
+    )
+    stage = process_stage.ProcessStage()
+    with pytest.raises(RuntimeError, match="initialization failed"):
+        await stage.initialize(
+            SimpleNamespace(
+                astrbot_config_id="profile-a", astrbot_config={"btw": {"enabled": True}}
+            )
+        )
+    assert runtime_registry._managers == {}
+    await stage.close()
+
+
+@pytest.mark.asyncio
+async def test_failing_close_still_removes_its_registration(monkeypatch):
+    monkeypatch.setattr(process_stage.AgentRequestSubStage, "initialize", AsyncMock())
+    monkeypatch.setattr(process_stage.StarRequestSubStage, "initialize", AsyncMock())
+    stage = process_stage.ProcessStage()
+    await stage.initialize(
+        SimpleNamespace(
+            astrbot_config_id="profile-a", astrbot_config={"btw": {"enabled": True}}
+        )
+    )
+    monkeypatch.setattr(
+        stage.conversation_loop,
+        "close",
+        AsyncMock(side_effect=RuntimeError("close failed")),
+    )
+    with pytest.raises(RuntimeError, match="close failed"):
+        await stage.close()
+    assert runtime_registry._managers == {}

tests/unit/test_builtin_command_extensions.py

@@ -170,11 +170,11 @@ async def test_work_submit_requires_both_loop_switches(config):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("task", ["", "status", " STATUS "])
-async def test_work_empty_and_status_remainders_show_usage(task):
+async def test_work_empty_and_status_remainders_query_status(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 _plain_text(event.result) == "No BTW work task has run in this session."
     assert event.is_stopped()
     assert event.get_extra("btw_force_work") is None
 

tests/unit/test_core_import_smoke.py

@@ -12,6 +12,10 @@ def test_btw_sdk_enable_check_does_not_construct_runtime(tmp_path: Path) -> None
 import pathlib
 import sys
 from astrbot.api import btw_work_loop_enabled
+from astrbot.api import btw_work_latest_status
+from astrbot.core.agent.btw import runtime_registry
+assert callable(btw_work_latest_status)
+assert runtime_registry._managers == {}
 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