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