fix(btw): record runner errors and refused work runs

astrbot/core/agent/btw/types.py

@@ -18,6 +18,27 @@ def is_work_loop_enabled(config: object) -> bool:
     return isinstance(work, Mapping) and bool(work.get("enabled", False))
 
 
+# How a run tells the work loop that it did not succeed.  The writers are the
+# agent stages and the Agent runner; the work loop is the only reader.  They
+# live here so a writer and its reader cannot drift apart: a run that fails
+# without setting one of these is reported as completed.
+WORK_FAILED_EXTRA = "btw_work_failed"
+THIRD_PARTY_RUNNER_ERROR_EXTRA_KEY = "_third_party_runner_error"
+
+
+def mark_work_run_failed(event) -> None:
+    """Record that a run failed, for work runs only.
+
+    Chat runs keep their own error path, so the marker is written only when the
+    event belongs to the work loop.
+
+    Args:
+        event: The event whose run ended in an error.
+    """
+    if event.get_extra("btw_loop") == "work":
+        event.set_extra(WORK_FAILED_EXTRA, True)
+
+
 class TaskType(StrEnum):
     """The execution loop selected for a user request."""
 

astrbot/core/agent/btw/work_loop.py

@@ -12,7 +12,12 @@
 from astrbot.core.utils.task_utils import create_tracked_task
 
 from . import i18n as work_i18n
-from .types import WorkSession, WorkSessionStatus
+from .types import (
+    THIRD_PARTY_RUNNER_ERROR_EXTRA_KEY,
+    WORK_FAILED_EXTRA,
+    WorkSession,
+    WorkSessionStatus,
+)
 from .work_sessions import WorkSessionManager
 
 
@@ -231,13 +236,15 @@ async def _execute(
         session_id: str,
     ) -> AsyncGenerator[None]:
         """Run one already-created work session and update its lifecycle."""
+        produced = False
         try:
             async with self._semaphore:
                 await self.sessions.update_status(
                     session_id,
                     WorkSessionStatus.RUNNING,
                 )
                 async for progress in self.executor.process(event):
+                    produced = True
                     yield progress
         except asyncio.CancelledError:
             await self.sessions.update_status(
@@ -253,13 +260,19 @@ async def _execute(
             )
             raise
         else:
-            failed = bool(event.get_extra("btw_work_failed"))
+            failed = bool(event.get_extra(WORK_FAILED_EXTRA)) or bool(
+                event.get_extra(THIRD_PARTY_RUNNER_ERROR_EXTRA_KEY)
+            )
             # An admitted stop request and a run that reported its own abort are
             # both cancellations.  ``run_agent`` clears ``agent_stop_requested``
             # when it reports the abort, so the stop flag alone misses that case.
             cancelled = bool(event.get_extra("agent_stop_requested")) or bool(
                 event.get_extra("agent_user_aborted")
             )
+            # An executor that produced nothing never reached an Agent: a run
+            # the session turned away is the reported case.  The generator
+            # ending only proves the task ran when something actually ran.
+            failed = failed or (not produced and not cancelled)
             await self.sessions.update_status(
                 session_id,
                 WorkSessionStatus.FAILED

astrbot/core/astr_agent_run_util.py

@@ -2,6 +2,7 @@
 from collections.abc import AsyncGenerator
 
 from astrbot import logger
+from astrbot.core.agent.btw.types import mark_work_run_failed
 from astrbot.core.agent.llm_types import LLMResponse
 from astrbot.core.agent.message import Message
 from astrbot.core.agent.runners.tool_loop_agent_runner import ToolLoopAgentRunner
@@ -334,6 +335,11 @@ async def run_agent(
                         astr_event.set_extra("agent_user_aborted", True)
                         astr_event.set_extra("agent_stop_requested", False)
                         return
+                    if resp.type == "err":
+                        # The run turns this into a user-facing error result and
+                        # ends normally, so the work loop cannot tell it apart
+                        # from a task that produced an answer.
+                        mark_work_run_failed(astr_event)
                     if _should_stop_agent(astr_event):
                         continue
 
@@ -367,6 +373,9 @@ async def run_agent(
             "Agent execution failed: %s",
             safe_error("", e),
         )
+        # The exception is reported to the user as an error result, so the run
+        # still ends by draining this generator normally.
+        mark_work_run_failed(astr_event)
         err_msg = get_agent_error_message(astr_event)
 
         error_llm_response = LLMResponse(

astrbot/core/pipeline/process_stage/method/agent_request.py

@@ -1,6 +1,7 @@
 from collections.abc import AsyncGenerator
 
 from astrbot import logger
+from astrbot.core.agent.btw.types import mark_work_run_failed
 from astrbot.core.config.agent_runner import normalize_agent_runner_for_load
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
 from astrbot.core.star.session_llm_manager import SessionServiceManager
@@ -33,12 +34,17 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
             logger.debug(
                 "This pipeline does not enable AI capability, skip processing."
             )
+            # A work run that is turned away here never reaches an Agent, so the
+            # generator ends without running anything.  Record that as a failure
+            # rather than letting it read as a completed task.
+            mark_work_run_failed(event)
             return
 
         if not await self.session_services.should_process_llm_request(event):
             logger.debug(
                 f"The session {event.unified_msg_origin} has disabled AI capability, skipping processing."
             )
+            mark_work_run_failed(event)
             return
 
         async for resp in self.agent_sub_stage.process(event):

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

@@ -6,6 +6,7 @@
 from dataclasses import replace
 
 from astrbot import logger
+from astrbot.core.agent.btw.types import WORK_FAILED_EXTRA
 from astrbot.core.agent.follow_up import FollowUpCapture
 from astrbot.core.agent.llm_types import (
     LLMResponse,
@@ -426,7 +427,7 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
                     )
                     if build_result is None:
                         if is_detached_work:
-                            event.set_extra("btw_work_failed", True)
+                            event.set_extra(WORK_FAILED_EXTRA, True)
                         return
 
                     agent_runner = build_result.agent_runner
@@ -570,7 +571,7 @@ async def process(self, event: AstrMessageEvent) -> AsyncGenerator[None]:
 
         except Exception as e:
             if is_detached_work:
-                event.set_extra("btw_work_failed", True)
+                event.set_extra(WORK_FAILED_EXTRA, True)
             logger.error(
                 "Error occurred while processing agent: %s",
                 safe_error("", e),

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

@@ -35,6 +35,7 @@
 if TYPE_CHECKING:
     from astrbot.core.agent.llm_types import LLMResponse
     from astrbot.core.agent.runners.base import BaseAgentRunner
+from astrbot.core.agent.btw.types import THIRD_PARTY_RUNNER_ERROR_EXTRA_KEY
 from astrbot.core.agent.llm_types import (
     ProviderRequest,
 )
@@ -47,7 +48,6 @@
 from .....astr_agent_context import AgentContextWrapper, AstrAgentContext
 from ....context import PipelineContext, call_event_hook
 
-THIRD_PARTY_RUNNER_ERROR_EXTRA_KEY = "_third_party_runner_error"
 STREAM_CONSUMPTION_CLOSE_TIMEOUT_SEC = 30
 RUNNER_NO_RESULT_FALLBACK_MESSAGE = DEFAULT_AGENT_ERROR_MESSAGE
 RUNNER_NO_FINAL_RESPONSE_LOG = (

tests/unit/test_agent_request_sub_stage.py

@@ -125,13 +125,21 @@ def __init__(
         self,
         unified_msg_origin: str = "umo-1",
         platform_name: str = "test",
+        extras: dict | None = None,
     ) -> None:
         self.unified_msg_origin = unified_msg_origin
         self.platform_name = platform_name
+        self.extras = extras or {}
 
     def get_platform_name(self) -> str:
         return self.platform_name
 
+    def get_extra(self, key: str):
+        return self.extras.get(key)
+
+    def set_extra(self, key: str, value) -> None:
+        self.extras[key] = value
+
 
 async def _yield_items(*items):
     for item in items:
@@ -215,11 +223,14 @@ async def test_process_returns_early_when_provider_is_disabled(monkeypatch):
         should_process,
     )
 
-    outputs = [item async for item in stage.process(FakeEvent())]
+    event = FakeEvent(extras={"btw_loop": "work"})
+    outputs = [item async for item in stage.process(event)]
 
     assert outputs == []
     should_process.assert_not_awaited()
     assert stage.agent_sub_stage.process_calls == []
+    # A work run turned away here never reached an Agent.
+    assert event.get_extra("btw_work_failed") is True
 
 
 @pytest.mark.asyncio
@@ -235,11 +246,33 @@ async def test_process_returns_early_when_session_llm_is_disabled(monkeypatch):
         should_process,
     )
 
-    outputs = [item async for item in stage.process(FakeEvent("umo-disabled"))]
+    event = FakeEvent("umo-disabled", extras={"btw_loop": "work"})
+    outputs = [item async for item in stage.process(event)]
 
     assert outputs == []
     should_process.assert_awaited_once()
     assert stage.agent_sub_stage.process_calls == []
+    assert event.get_extra("btw_work_failed") is True
+
+
+@pytest.mark.asyncio
+async def test_process_leaves_a_refused_chat_run_unmarked(monkeypatch):
+    """Only the work loop reads the failure marker; chat runs keep their path."""
+    stage = agent_request.AgentRequestSubStage()
+    ctx = _ctx()
+    await stage.initialize(ctx)
+
+    monkeypatch.setattr(
+        agent_request.SessionServiceManager,
+        "should_process_llm_request",
+        AsyncMock(return_value=False),
+    )
+    event = FakeEvent("umo-disabled")
+
+    outputs = [item async for item in stage.process(event)]
+
+    assert outputs == []
+    assert event.get_extra("btw_work_failed") is None
 
 
 @pytest.mark.asyncio

tests/unit/test_btw_work_loop.py

@@ -9,6 +9,10 @@
 from astrbot.core.agent.btw.work_loop import WorkLoop
 from astrbot.core.agent.btw.work_sessions import WorkSessionManager
 from astrbot.core.astr_agent_run_util import run_agent
+from astrbot.core.message.message_event_result import MessageChain
+from astrbot.core.pipeline.process_stage.method import agent_request
+from astrbot.core.pipeline.process_stage.method.agent_sub_stages import third_party
+from tests.unit.agent_sub_stage_support import FakeThirdPartyRunner
 from tests.unit.test_astr_agent_run_util import FakeEvent as RunnerEvent
 from tests.unit.test_astr_agent_run_util import FakeRunner
 
@@ -81,6 +85,45 @@ async def process(self, event):
             yield
 
 
+class ThirdPartyExecutor:
+    """The real third-party response handler as the work loop's executor."""
+
+    def __init__(self, runner) -> None:
+        self.runner = runner
+
+    async def process(self, event):
+        stage = third_party.ThirdPartyAgentSubStage.__new__(
+            third_party.ThirdPartyAgentSubStage
+        )
+        async for _ in stage._handle_non_streaming_response(
+            runner=self.runner,
+            event=event,
+            stream_to_general=False,
+            custom_error_message=None,
+        ):
+            yield
+
+
+class AdmissionSkipExecutor:
+    """The real Agent request stage for a run the session turns away."""
+
+    def __init__(self, stage) -> None:
+        self.stage = stage
+
+    async def process(self, event):
+        async for progress in self.stage.process(event):
+            yield progress
+
+
+class SilentExecutor:
+    """An executor that ends without running anything or reporting a result."""
+
+    async def process(self, event):
+        del event
+        if False:  # noqa: SIM223 -- unreachable yield keeps this a generator
+            yield
+
+
 @pytest.mark.asyncio
 async def test_work_loop_marks_failures_without_exposing_executor_error():
     sessions = WorkSessionManager()
@@ -283,3 +326,91 @@ async def test_work_loop_records_user_abort_as_cancelled():
     session = await sessions.get_for_origin(event.unified_msg_origin)
     assert session is not None
     assert session.status is WorkSessionStatus.CANCELLED
+
+
+@pytest.mark.asyncio
+async def test_work_loop_records_agent_error_response_as_failed():
+    """A local ``err`` response ends the generator normally but is a failure."""
+    event = WorkEvent()
+    runner = FakeRunner(
+        [SimpleNamespace(type="err", data={"chain": MessageChain().message("boom")})],
+        event=event,
+    )
+    sessions = WorkSessionManager()
+    work_loop = WorkLoop(RunAgentExecutor(runner), sessions)
+
+    _ = [item async for item in work_loop.process(event)]
+
+    session = await sessions.get_for_origin(event.unified_msg_origin)
+    assert session is not None
+    assert session.status is WorkSessionStatus.FAILED
+
+
+@pytest.mark.asyncio
+async def test_work_loop_records_runner_exception_as_failed():
+    """run_agent converts a raising runner into a result, not into a re-raise."""
+    event = WorkEvent()
+    runner = FakeRunner(RuntimeError("provider exploded"), event=event, streaming=True)
+    sessions = WorkSessionManager()
+    work_loop = WorkLoop(RunAgentExecutor(runner), sessions)
+
+    _ = [item async for item in work_loop.process(event)]
+
+    session = await sessions.get_for_origin(event.unified_msg_origin)
+    assert session is not None
+    assert session.status is WorkSessionStatus.FAILED
+
+
+@pytest.mark.asyncio
+async def test_work_loop_records_third_party_runner_error_as_failed():
+    """A third-party runner failure is reported through its own marker."""
+    event = WorkEvent()
+    runner = FakeThirdPartyRunner(
+        step_exception=RuntimeError("service unavailable"),
+        final_resp=None,
+    )
+    sessions = WorkSessionManager()
+    work_loop = WorkLoop(ThirdPartyExecutor(runner), sessions)
+
+    _ = [item async for item in work_loop.process(event)]
+
+    assert event.get_extra(third_party.THIRD_PARTY_RUNNER_ERROR_EXTRA_KEY) is True
+    session = await sessions.get_for_origin(event.unified_msg_origin)
+    assert session is not None
+    assert session.status is WorkSessionStatus.FAILED
+
+
+@pytest.mark.asyncio
+async def test_work_loop_records_a_run_the_session_refused_as_failed():
+    """Session admission can decline a work run before any Agent is built."""
+    event = WorkEvent()
+    stage = agent_request.AgentRequestSubStage.__new__(
+        agent_request.AgentRequestSubStage
+    )
+    stage.ctx = SimpleNamespace(astrbot_config={"provider_settings": {"enable": True}})
+    stage.session_services = SimpleNamespace(
+        should_process_llm_request=AsyncMock(return_value=False)
+    )
+    sessions = WorkSessionManager()
+    work_loop = WorkLoop(AdmissionSkipExecutor(stage), sessions)
+
+    _ = [item async for item in work_loop.process(event)]
+
+    assert event.get_extra("btw_loop") == "work"
+    session = await sessions.get_for_origin(event.unified_msg_origin)
+    assert session is not None
+    assert session.status is WorkSessionStatus.FAILED
+
+
+@pytest.mark.asyncio
+async def test_work_loop_does_not_report_a_silent_run_as_completed():
+    """An executor that ran nothing is not evidence the task succeeded."""
+    event = FakeEvent("执行命令")
+    sessions = WorkSessionManager()
+    work_loop = WorkLoop(SilentExecutor(), sessions)
+
+    _ = [item async for item in work_loop.process(event)]
+
+    session = await sessions.get_for_origin(event.unified_msg_origin)
+    assert session is not None
+    assert session.status is WorkSessionStatus.FAILED