From bec25496ea9086b73f603bd78a4ec80519598ad0 Mon Sep 17 00:00:00 2001 From: YUZHEthefool <2804776511@qq.com> Date: Fri, 11 Sep 2026 21:48:32 +0800 Subject: [PATCH] fix(btw): record runner errors and refused work runs The work loop read a single failure marker, so three failing paths ended as completed: - run_agent reports a provider error as a user-facing result and then drains its generator normally, which is indistinguishable from an answer. - The third-party stage records _third_party_runner_error and nothing read it. - A run the session refuses on admission never reaches an Agent, so its executor yields nothing and the generator ending looked like success. Write the failure marker from each of those paths and read both markers in the work loop. The marker keys move into btw/types.py so the writers and their only reader cannot drift apart; mark_work_run_failed keeps chat runs on their existing error path. An executor that produced nothing also no longer counts as completed: a generator ending only proves the task ran when something actually ran. Fixes #154 AI-Generated: true Generated-At: 2026-09-11T13:48:32Z --- astrbot/core/agent/btw/types.py | 21 +++ astrbot/core/agent/btw/work_loop.py | 17 ++- astrbot/core/astr_agent_run_util.py | 9 ++ .../process_stage/method/agent_request.py | 6 + .../method/agent_sub_stages/internal.py | 5 +- .../method/agent_sub_stages/third_party.py | 2 +- tests/unit/test_agent_request_sub_stage.py | 37 ++++- tests/unit/test_btw_work_loop.py | 131 ++++++++++++++++++ 8 files changed, 221 insertions(+), 7 deletions(-) diff --git a/astrbot/core/agent/btw/types.py b/astrbot/core/agent/btw/types.py index ca600b8d87..6bba6a936b 100644 --- a/astrbot/core/agent/btw/types.py +++ b/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.""" diff --git a/astrbot/core/agent/btw/work_loop.py b/astrbot/core/agent/btw/work_loop.py index dd9f916fc0..36312c9713 100644 --- a/astrbot/core/agent/btw/work_loop.py +++ b/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,6 +236,7 @@ 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( @@ -238,6 +244,7 @@ async def _execute( 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 diff --git a/astrbot/core/astr_agent_run_util.py b/astrbot/core/astr_agent_run_util.py index f3aad31d7a..f9d5e1db67 100644 --- a/astrbot/core/astr_agent_run_util.py +++ b/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( diff --git a/astrbot/core/pipeline/process_stage/method/agent_request.py b/astrbot/core/pipeline/process_stage/method/agent_request.py index 3754c5e433..a9ba262395 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_request.py +++ b/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): diff --git a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py index cc8fe618f8..4d348efd75 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py +++ b/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), diff --git a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py b/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py index 5e1e4d4f79..5a9b557974 100644 --- a/astrbot/core/pipeline/process_stage/method/agent_sub_stages/third_party.py +++ b/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 = ( diff --git a/tests/unit/test_agent_request_sub_stage.py b/tests/unit/test_agent_request_sub_stage.py index e239067593..457d2fe90e 100644 --- a/tests/unit/test_agent_request_sub_stage.py +++ b/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 diff --git a/tests/unit/test_btw_work_loop.py b/tests/unit/test_btw_work_loop.py index 4782793ab4..d1457d19e9 100644 --- a/tests/unit/test_btw_work_loop.py +++ b/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