feat(btw): deliver background results to the originating request

astrbot/core/agent/btw/work_loop.py

@@ -2,6 +2,7 @@
 
 import asyncio
 from collections.abc import AsyncGenerator, Awaitable, Callable
+from contextlib import aclosing
 from typing import Protocol
 
 from astrbot import logger
@@ -50,6 +51,8 @@ def __init__(
         self._background_tasks: set[asyncio.Task] | None = None
         self._result_dispatcher: ResultDispatcher | None = None
         self._event_finalizer: EventFinalizer | None = None
+        self._tasks: dict[asyncio.Task, tuple[AstrMessageEvent, str]] = {}
+        self._closed = False
 
     def configure_detached_execution(
         self,
@@ -92,6 +95,17 @@ async def submit(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
         Falls back to inline execution when no runtime task registry is
         attached, which keeps the primitive usable in isolated tests.
         """
+        if self._closed:
+            event.set_result(
+                MessageEventResult().message(
+                    work_i18n.text(
+                        work_i18n.resolve_event_locale(event),
+                        "btw.work.status.cancelled",
+                    )
+                )
+            )
+            yield
+            return
         if (
             self._background_tasks is None
             or self._result_dispatcher is None
@@ -114,15 +128,43 @@ async def submit(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
         )
         yield
 
+        if self._closed:
+            await self.sessions.update_status(session.id, WorkSessionStatus.CANCELLED)
+            return
+
         # 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(
+        task = create_tracked_task(
             self._background_tasks,
             self._run_detached(event, session.id),
             name=f"btw_work:{session.id}",
         )
+        self._tasks[task] = (event, session.id)
+        task.add_done_callback(lambda done: self._tasks.pop(done, None))
+
+    async def close(self) -> None:
+        """Cancel and finalize this profile's work, including unstarted tasks."""
+        self._closed = True
+        tasks = dict(self._tasks)
+        for task in tasks:
+            task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        finalizers = []
+        if self._event_finalizer is not None:
+            for event, session_id in tasks.values():
+                if not event.get_extra("btw_detached_work_finished"):
+                    await self.sessions.update_status(
+                        session_id, WorkSessionStatus.CANCELLED
+                    )
+                    finalizers.append(self._event_finalizer(event))
+        if finalizers:
+            results = await asyncio.gather(*finalizers, return_exceptions=True)
+            for result in results:
+                if isinstance(result, BaseException):
+                    raise result
 
     @staticmethod
     def _prepare_event(event: AstrMessageEvent, session_id: str) -> None:
@@ -159,21 +201,35 @@ async def _execute(
             )
             raise
         else:
+            failed = bool(event.get_extra("btw_work_failed"))
+            cancelled = bool(event.get_extra("agent_stop_requested"))
             await self.sessions.update_status(
                 session_id,
-                WorkSessionStatus.COMPLETED,
+                WorkSessionStatus.FAILED
+                if failed
+                else (
+                    WorkSessionStatus.CANCELLED
+                    if cancelled
+                    else WorkSessionStatus.COMPLETED
+                ),
+                error="Work task failed." if failed else None,
             )
 
     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)
+            async with aclosing(self._execute(event, session_id)) as execution:
+                async for _ in execution:
+                    await self._result_dispatcher(event)
         except asyncio.CancelledError:
+            await self.sessions.update_status(session_id, WorkSessionStatus.CANCELLED)
             raise
         except Exception as exc:
+            await self.sessions.update_status(
+                session_id, WorkSessionStatus.FAILED, error="Work task failed."
+            )
             # 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))

astrbot/core/agent/conversation_loop.py

@@ -1,6 +1,7 @@
 """Opt-in conversation entry over the existing Agent request executor."""
 
-from collections.abc import AsyncGenerator
+import asyncio
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import TYPE_CHECKING
 
 from astrbot.core.agent.btw.types import is_work_loop_enabled
@@ -43,6 +44,27 @@ async def initialize(self, ctx: PipelineContext) -> None:
             max_concurrent=concurrency if type(concurrency) is int else 2,
         )
 
+    def configure_detached_work(
+        self,
+        *,
+        background_tasks: set[asyncio.Task],
+        result_dispatcher: Callable[[AstrMessageEvent], Awaitable[None]],
+        event_finalizer: Callable[[AstrMessageEvent], Awaitable[None]],
+    ) -> None:
+        """Attach the owning scheduler's delivery and cleanup services."""
+        if self.work_loop is None:
+            raise RuntimeError("ConversationLoop is not initialized")
+        self.work_loop.configure_detached_execution(
+            background_tasks=background_tasks,
+            result_dispatcher=result_dispatcher,
+            event_finalizer=event_finalizer,
+        )
+
+    async def close(self) -> None:
+        """Stop work owned by this conversation entry."""
+        if self.work_loop is not None:
+            await self.work_loop.close()
+
     async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
         """Process one admitted conversation using the current Agent path."""
         if (

astrbot/core/core_lifecycle.py

@@ -723,6 +723,8 @@ async def stop(self) -> None:
 
             if not self._cleanup_stack_closed:
                 self._cleanup_stack_closed = True
+                for scheduler in self.pipeline_scheduler_mapping.values():
+                    self._register_cleanup("pipeline work tasks", scheduler.close)
                 await self._cleanup_stack.aclose()
 
             self._initialized = False
@@ -830,6 +832,9 @@ async def reload_pipeline_scheduler(self, conf_id: str) -> None:
                 getattr(self.services, "authorization", None),
             ),
         )
+        old_scheduler = self.pipeline_scheduler_mapping.get(conf_id)
+        if old_scheduler is not None:
+            await old_scheduler.close()
         await scheduler.initialize()
         self.pipeline_scheduler_mapping[conf_id] = scheduler
         manager = getattr(self, "turn_window_manager", None)
@@ -838,4 +843,6 @@ async def reload_pipeline_scheduler(self, conf_id: str) -> None:
 
     async def remove_pipeline_scheduler(self, conf_id: str) -> None:
         """Remove the scheduler associated with a deleted configuration profile."""
-        self.pipeline_scheduler_mapping.pop(conf_id, None)
+        scheduler = self.pipeline_scheduler_mapping.pop(conf_id, None)
+        if scheduler is not None:
+            await scheduler.close()

astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py

@@ -300,6 +300,7 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
         follow_up_consumed_marked = False
         follow_up_activated = False
         typing_requested = False
+        is_detached_work = bool(event.get_extra("btw_detached_work"))
         try:
             from astrbot.core.streaming_override import resolve_streaming_response
 
@@ -356,7 +357,9 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
 
             logger.debug("ready to request llm provider")
             follow_up_capture = (
-                self.ctx.execution_context.follow_up_coordinator.try_capture(event)
+                None
+                if is_detached_work
+                else self.ctx.execution_context.follow_up_coordinator.try_capture(event)
             )
             if follow_up_capture:
                 (
@@ -393,6 +396,9 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
             concurrent, lock_key, turn_cm, streaming_response = (
                 self._prepare_group_sender_concurrency(event, streaming_response)
             )
+            work_lock = event.get_extra("btw_agent_lock_key")
+            if is_detached_work and isinstance(work_lock, str) and work_lock:
+                lock_key = work_lock
 
             async with (
                 turn_cm,
@@ -409,6 +415,8 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
                         streaming_response,
                     )
                     if build_result is None:
+                        if is_detached_work:
+                            event.set_extra("btw_work_failed", True)
                         return
 
                     agent_runner = build_result.agent_runner
@@ -469,8 +477,9 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
                             )
                         else:
                             runner_stop_callback = None
-                    self._register_follow_up_runner(event, agent_runner, concurrent)
-                    runner_registered = True
+                    if not is_detached_work:
+                        self._register_follow_up_runner(event, agent_runner, concurrent)
+                        runner_registered = True
                     event.trace.record(
                         "astr_agent_prepare",
                         system_prompt=req.system_prompt,
@@ -550,6 +559,8 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
                         )
 
         except Exception as e:
+            if is_detached_work:
+                event.set_extra("btw_work_failed", True)
             logger.error(
                 "Error occurred while processing agent: %s",
                 safe_error("", e),

astrbot/core/pipeline/process_stage/stage.py

@@ -1,4 +1,5 @@
-from collections.abc import AsyncGenerator
+import asyncio
+from collections.abc import AsyncGenerator, Awaitable, Callable
 
 from astrbot.core.agent.conversation_loop import ConversationLoop
 from astrbot.core.agent.llm_types import ProviderRequest
@@ -29,6 +30,26 @@ async def initialize(self, ctx: PipelineContext) -> None:
         self.star_request_sub_stage = StarRequestSubStage()
         await self.star_request_sub_stage.initialize(ctx)
 
+    def configure_detached_work(
+        self,
+        *,
+        background_tasks: set[asyncio.Task],
+        result_dispatcher: Callable[[AstrMessageEvent], Awaitable[None]],
+        event_finalizer: Callable[[AstrMessageEvent], Awaitable[None]],
+    ) -> None:
+        """Attach runtime services only to an enabled conversation loop."""
+        if self.conversation_loop is not None:
+            self.conversation_loop.configure_detached_work(
+                background_tasks=background_tasks,
+                result_dispatcher=result_dispatcher,
+                event_finalizer=event_finalizer,
+            )
+
+    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()
+
     async def process(
         self,
         event: AstrMessageEvent,

astrbot/core/pipeline/scheduler.py

@@ -8,6 +8,7 @@
 
 from .bootstrap import builtin_stage_classes
 from .context import PipelineContext
+from .result_decorate.stage import ResultDecorateStage
 from .stage import Stage
 
 
@@ -33,6 +34,50 @@ async def initialize(self) -> None:
             stage_instance = stage_cls()  # 创建实例
             await stage_instance.initialize(self.ctx)
             self.stages.append(stage_instance)
+        for stage in self.stages:
+            configure = getattr(stage, "configure_detached_work", None)
+            if callable(configure):
+                configure(
+                    background_tasks=self.ctx.execution_context.background_tasks,
+                    result_dispatcher=self.deliver_detached_result,
+                    event_finalizer=self.finalize_detached_event,
+                )
+
+    async def close(self) -> None:
+        """Close profile-owned background work before replacing this scheduler."""
+        for stage in reversed(self.stages):
+            close = getattr(stage, "close", None)
+            if callable(close):
+                await cast(Awaitable[None], close())
+
+    async def deliver_detached_result(self, event: AstrMessageEvent) -> None:
+        """Replay response decoration and delivery with onion ordering intact."""
+        index = next(
+            (
+                i
+                for i, stage in enumerate(self.stages)
+                if isinstance(stage, ResultDecorateStage)
+            ),
+            None,
+        )
+        if index is None:
+            raise RuntimeError("ResultDecorateStage is not configured")
+        if not event.is_stopped():
+            await self._process_stages(event, index)
+
+    async def finalize_detached_event(self, event: AstrMessageEvent) -> None:
+        """Complete a retained request once, then release its resources."""
+        if event.get_extra("btw_detached_work_finished"):
+            return
+        event.set_extra("btw_detached_work_finished", True)
+        try:
+            if event.requires_empty_completion and not event.get_extra(
+                "skip_empty_completion"
+            ):
+                await cast(_EmptyCompletionEvent, event).send(None)
+        finally:
+            event.cleanup_temporary_local_files()
+            self.ctx.execution_context.active_event_registry.unregister(event)
 
     async def _process_stages(self, event: AstrMessageEvent, from_stage=0) -> None:
         """依次执行各个阶段
@@ -91,8 +136,10 @@ async def execute(self, event: AstrMessageEvent) -> None:
             await self._process_stages(event)
 
             # 发送一个空消息, 以便于后续的处理
-            if event.requires_empty_completion and not event.get_extra(
-                "skip_empty_completion"
+            if (
+                event.requires_empty_completion
+                and not event.get_extra("skip_empty_completion")
+                and not event.get_extra("btw_detached_work")
             ):
                 # Only adapters whose send implementation accepts ``None`` set this
                 # flag. The base event contract deliberately remains message-only.
@@ -112,5 +159,6 @@ async def execute(self, event: AstrMessageEvent) -> None:
             else:
                 logger.debug("pipeline execution completed.")
         finally:
-            event.cleanup_temporary_local_files()
-            self.ctx.execution_context.active_event_registry.unregister(event)
+            if not event.get_extra("btw_detached_work"):
+                event.cleanup_temporary_local_files()
+                self.ctx.execution_context.active_event_registry.unregister(event)

docs/en/dev/astrbot-config.md

@@ -207,6 +207,8 @@ Automatic classifier candidates are evaluated separately. Enabling this entry do
 
 The work executor additionally requires `btw.work_loop.enabled`, also `false` by default. It reuses the Agent executor and records pending, running, completed, failed, and cancelled task states. `btw.work_loop.max_concurrent` limits active execution (default `2`); it does not impose a waiting-queue length limit. `btw.work_session.max_age_seconds` retains terminal states for `3600` seconds by default; active tasks do not expire, and expired terminal records are removed during the next session operation. Runtime-owned background services perform task execution and cleanup when attached by the scheduler.
 
+Detached work acknowledges receipt before execution and returns results through the current response-decoration and delivery stages, including reply content checks. Inbound stages are not rerun. WebChat keeps the original request identifier open through the final result; acknowledgement does not end the request. Temporary event files remain available to the worker and are released on completion, failure, or cancellation. Replacing or removing a profile cancels its owned work; runtime shutdown also reclaims it.
+
 ## WebUI and authentication
 
 Important `dashboard` defaults:

docs/zh/dev/astrbot-config.md

@@ -209,6 +209,8 @@ Alkaid [长期记忆](../use/long-term-memory) 当前没有对应的启停配置
 
 工作执行器还需要开启 `btw.work_loop.enabled`，默认同样为 `false`。它复用 Agent 执行器并记录排队、运行、完成、失败、取消状态。`btw.work_loop.max_concurrent` 限制正在执行的任务数，默认 `2`，不限制等待队列长度。`btw.work_session.max_age_seconds` 默认保留终态记录 `3600` 秒；活动任务不会过期，终态过期记录在下次会话操作时清除。调度器接入后台服务后，由运行时拥有工作任务的执行和清理。
 
+后台工作在执行前确认接收，再通过当前回复装饰与发送阶段回送结果，包括回复内容检查；不重复运行入站阶段。WebChat 持续使用原请求标识，确认消息不会结束请求。事件临时文件保留到工作完成、失败或取消后再释放。配置档替换、删除以及运行时关闭会取消并回收其工作任务。
+
 ## WebUI 与认证
 
 `dashboard` 的关键默认值：