feat(btw): own detached work execution and lifecycle

astrbot/core/agent/btw/__init__.py

@@ -0,0 +1 @@
+# BTW runtime primitives; keep package imports inert.

astrbot/core/agent/btw/i18n.py

@@ -0,0 +1,50 @@
+"""Locale-aware user-facing strings for the BTW work loop.
+
+The work loop runs in core without a plugin context, so it resolves the
+locale from the event extra/session the same way ``PluginContext._locale``
+does, then looks the string up in these bundles. Missing locales fall back
+to ``zh-CN``.
+"""
+
+LOCALES: dict[str, dict[str, str]] = {
+    "zh-CN": {
+        "btw.work.started": "🔧 工作任务已开始处理。",
+        "btw.work.status.pending": "工作任务正在排队。",
+        "btw.work.status.running": "工作任务正在执行。",
+        "btw.work.status.completed": "工作任务已完成。",
+        "btw.work.status.failed": "工作任务执行失败。",
+        "btw.work.status.cancelled": "工作任务已取消。",
+    },
+    "en-US": {
+        "btw.work.started": "🔧 Work task started.",
+        "btw.work.status.pending": "The work task is queued.",
+        "btw.work.status.running": "The work task is running.",
+        "btw.work.status.completed": "The work task is completed.",
+        "btw.work.status.failed": "The work task failed.",
+        "btw.work.status.cancelled": "The work task was cancelled.",
+    },
+}
+
+_FALLBACK_LOCALE = "zh-CN"
+
+
+def resolve_event_locale(event) -> str:
+    """Return the locale for an event (extra first, then the stored session)."""
+    getter = getattr(event, "get_extra", None)
+    if callable(getter):
+        try:
+            extra = getter("locale")
+        except Exception:  # noqa: BLE001
+            extra = None
+        if extra:
+            return str(extra)
+    return _FALLBACK_LOCALE
+
+
+def text(locale: str, key: str) -> str:
+    """Return one BTW string for a locale, falling back to zh-CN then key."""
+    bundle = LOCALES.get(locale) or LOCALES[_FALLBACK_LOCALE]
+    value = bundle.get(key)
+    if value is None:
+        value = LOCALES[_FALLBACK_LOCALE].get(key, key)
+    return value

astrbot/core/agent/btw/types.py

@@ -0,0 +1,65 @@
+"""Types shared by the BTW conversation and work loops."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from enum import StrEnum
+from uuid import uuid4
+
+
+def is_work_loop_enabled(config: object) -> bool:
+    """Return whether the profile explicitly enables BTW and work."""
+    if not isinstance(config, Mapping):
+        return False
+    btw = config.get("btw", {})
+    if not isinstance(btw, Mapping) or not btw.get("enabled", False):
+        return False
+    work = btw.get("work_loop", {})
+    return isinstance(work, Mapping) and bool(work.get("enabled", False))
+
+
+class TaskType(StrEnum):
+    """The execution loop selected for a user request."""
+
+    CONVERSATION = "conversation"
+    WORK = "work"
+
+
+class WorkSessionStatus(StrEnum):
+    """Lifecycle states for one work-loop request."""
+
+    PENDING = "pending"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+@dataclass(slots=True)
+class WorkSession:
+    """Runtime state shared by the conversation and work loops."""
+
+    origin: str
+    request: str
+    task_type: TaskType = TaskType.WORK
+    id: str = field(default_factory=lambda: uuid4().hex)
+    status: WorkSessionStatus = WorkSessionStatus.PENDING
+    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+    updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+    error: str | None = None
+
+    def update_status(
+        self,
+        status: WorkSessionStatus,
+        *,
+        error: str | None = None,
+    ) -> None:
+        """Record a status transition.
+
+        Args:
+            status: The new work-session status.
+            error: A safe diagnostic for failed work, when available.
+        """
+        self.status = status
+        self.error = error
+        self.updated_at = datetime.now(UTC)

astrbot/core/agent/btw/work_loop.py

@@ -0,0 +1,181 @@
+"""The BTW work-loop prototype backed by the existing Agent tool loop."""
+
+import asyncio
+from collections.abc import AsyncGenerator, Awaitable, Callable
+from typing import Protocol
+
+from astrbot import logger
+from astrbot.core.message.message_event_result import MessageEventResult
+from astrbot.core.platform.astr_message_event import AstrMessageEvent
+from astrbot.core.utils.error_redaction import safe_error
+from astrbot.core.utils.task_utils import create_tracked_task
+
+from . import i18n as work_i18n
+from .types import WorkSessionStatus
+from .work_sessions import WorkSessionManager
+
+
+class AgentRequestExecutor(Protocol):
+    """The existing Agent request path required by the work loop."""
+
+    def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
+        """Yield pipeline progress markers for one event.
+
+        Protocol stub; concrete implementations are the pipeline's Agent
+        request sub-stage.  The body raises so the statement is effectful
+        (CodeQL py/ineffectual-statement); the unreachable ``yield`` keeps
+        the declared ``AsyncGenerator`` return type type-checkable.
+        """
+        raise NotImplementedError
+        yield  # noqa: B901 -- unreachable marker for the type checker
+
+
+ResultDispatcher = Callable[[AstrMessageEvent], Awaitable[None]]
+EventFinalizer = Callable[[AstrMessageEvent], Awaitable[None]]
+
+
+class WorkLoop:
+    """Run classified work with the current Agent and tool infrastructure."""
+
+    def __init__(
+        self,
+        executor: AgentRequestExecutor,
+        sessions: WorkSessionManager,
+        *,
+        max_concurrent: int = 2,
+    ) -> None:
+        self.executor = executor
+        self.sessions = sessions
+        self._semaphore = asyncio.Semaphore(max(1, max_concurrent))
+        self._background_tasks: set[asyncio.Task] | None = None
+        self._result_dispatcher: ResultDispatcher | None = None
+        self._event_finalizer: EventFinalizer | None = None
+
+    def configure_detached_execution(
+        self,
+        *,
+        background_tasks: set[asyncio.Task],
+        result_dispatcher: ResultDispatcher,
+        event_finalizer: EventFinalizer,
+    ) -> None:
+        """Attach runtime-owned background execution services.
+
+        Args:
+            background_tasks: Runtime task registry cancelled during shutdown.
+            result_dispatcher: Delivers a generated work result through the
+                configured result-decorate and response stages.
+            event_finalizer: Releases the event after detached work finishes.
+        """
+        self._background_tasks = background_tasks
+        self._result_dispatcher = result_dispatcher
+        self._event_finalizer = event_finalizer
+
+    async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
+        """Execute one work-loop request inline.
+
+        Args:
+            event: The classified message event.
+
+        Yields:
+            Pipeline progress markers emitted by the existing Agent executor.
+        """
+        session = await self.sessions.create(
+            event.unified_msg_origin, event.message_str
+        )
+        self._prepare_event(event, session.id)
+        async for progress in self._execute(event, session.id):
+            yield progress
+
+    async def submit(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
+        """Acknowledge work, then run it without retaining the request pipeline.
+
+        Falls back to inline execution when no runtime task registry is
+        attached, which keeps the primitive usable in isolated tests.
+        """
+        if (
+            self._background_tasks is None
+            or self._result_dispatcher is None
+            or self._event_finalizer is None
+        ):
+            async for progress in self.process(event):
+                yield progress
+            return
+
+        session = await self.sessions.create(
+            event.unified_msg_origin, event.message_str
+        )
+        self._prepare_event(event, session.id)
+        event.set_result(
+            MessageEventResult().message(
+                work_i18n.text(
+                    work_i18n.resolve_event_locale(event), "btw.work.started"
+                )
+            )
+        )
+        yield
+
+        # The first yield returns only after the normal response stages deliver
+        # the acknowledgement.  Marking it here prevents the scheduler from
+        # releasing event-owned temporary files before the worker needs them.
+        event.set_extra("btw_detached_work", True)
+        create_tracked_task(
+            self._background_tasks,
+            self._run_detached(event, session.id),
+            name=f"btw_work:{session.id}",
+        )
+
+    @staticmethod
+    def _prepare_event(event: AstrMessageEvent, session_id: str) -> None:
+        """Mark an event so Agent assembly uses the work-loop policy."""
+        event.set_extra("btw_work_session_id", session_id)
+        event.set_extra("btw_loop", "work")
+        event.set_extra("btw_agent_lock_key", f"{event.unified_msg_origin}:work")
+
+    async def _execute(
+        self,
+        event: AstrMessageEvent,
+        session_id: str,
+    ) -> AsyncGenerator[None]:
+        """Run one already-created work session and update its lifecycle."""
+        try:
+            async with self._semaphore:
+                await self.sessions.update_status(
+                    session_id,
+                    WorkSessionStatus.RUNNING,
+                )
+                async for progress in self.executor.process(event):
+                    yield progress
+        except asyncio.CancelledError:
+            await self.sessions.update_status(
+                session_id,
+                WorkSessionStatus.CANCELLED,
+            )
+            raise
+        except Exception:
+            await self.sessions.update_status(
+                session_id,
+                WorkSessionStatus.FAILED,
+                error="Work task failed.",
+            )
+            raise
+        else:
+            await self.sessions.update_status(
+                session_id,
+                WorkSessionStatus.COMPLETED,
+            )
+
+    async def _run_detached(self, event: AstrMessageEvent, session_id: str) -> None:
+        """Run work in the runtime task registry and deliver each result."""
+        assert self._result_dispatcher is not None
+        assert self._event_finalizer is not None
+        try:
+            async for _ in self._execute(event, session_id):
+                await self._result_dispatcher(event)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            # The task registry logs unhandled exceptions with their traceback.
+            # Consume executor failures here so provider details never reach it.
+            logger.error("BTW work task failed: %s", safe_error("", exc))
+        finally:
+            await self._event_finalizer(event)