From 59cc1daee191eeff97666d261b35071afd431b18 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Sat, 18 Jul 2026 03:42:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9EAgentRun=E7=BB=88?= =?UTF-8?q?=E6=80=81=E6=84=9F=E7=9F=A5=E8=83=BD=E5=8A=9B=E4=B8=8E=E9=85=8D?= =?UTF-8?q?=E5=A5=97=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本次提交实现了AgentRun全生命周期终态处理能力，覆盖以下核心变更： 1. 新增4种AgentRun终态错误类型与HTTP状态映射，透传上游错误上下文 2. 新增AgentRun终态信息查询接口与DTO，支持非事件流场景下的状态感知 3. 优化沙箱路径转义规则，兼容服务账号UID格式 4. 新增出站管道SKIP逻辑，针对取消/中断场景避免无效重试 5. 补充完整单元测试覆盖所有新功能与优化点 --- .../yuxi/agents/backends/sandbox/paths.py | 22 +- .../channels/adapters/agent_run_adapter.py | 59 +++- .../adapters/agent_run_execution_adapter.py | 13 + .../inbound/agent_run_enqueue_stage.py | 30 +- .../pipeline/outbound/load_build_stage.py | 88 ++++- .../usecase/inbound_message_service.py | 24 ++ .../contract/dtos/shared/agent_run.py | 26 ++ .../yuxi/channels/contract/errors/__init__.py | 9 + .../channels/contract/errors/_status_map.py | 11 + .../contract/errors/domain/__init__.py | 8 + .../channels/contract/errors/domain/agent.py | 152 +++++++++ .../channels/contract/policy/config_schema.py | 18 +- .../driven/shared/agent_run_execution_port.py | 28 ++ .../ports/driven/shared/agent_run_port.py | 46 ++- .../yuxi/services/agent_run_service.py | 27 ++ .../test/unit/backends/test_sandbox_paths.py | 44 +++ .../test_sandbox_provisioner_config.py | 11 +- .../adapters/test_agent_run_adapter.py | 276 +++++++++++++++- .../outbound/test_load_build_stage.py | 302 +++++++++++++++++- .../usecase/test_inbound_message_service.py | 256 +++++++++++++++ 20 files changed, 1414 insertions(+), 36 deletions(-) create mode 100644 backend/test/unit/backends/test_sandbox_paths.py diff --git a/backend/package/yuxi/agents/backends/sandbox/paths.py b/backend/package/yuxi/agents/backends/sandbox/paths.py index 64746628..3757fea7 100644 --- a/backend/package/yuxi/agents/backends/sandbox/paths.py +++ b/backend/package/yuxi/agents/backends/sandbox/paths.py @@ -15,6 +15,9 @@ from yuxi.utils.paths import ( ) _SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$") +# 与 docker/sandbox_provisioner/app.py 中的 UNSAFE_PATH_SEGMENT_CHAR_RE 保持一致， +# 确保宿主机路径与容器挂载路径采用相同的转义规则。 +_UNSAFE_PATH_SEGMENT_CHAR_RE = re.compile(r"[^A-Za-z0-9_-]") def get_virtual_path_prefix() -> str: @@ -35,18 +38,23 @@ def _thread_root_dir(thread_id: str) -> Path: return Path(conf.save_dir) / "threads" / safe_thread_id / "user-data" -def _validate_uid(uid: str) -> str: - value = str(uid or "").strip() - if not value: +def _sanitize_uid(uid: str) -> str: + """将 uid 转义为文件系统安全的路径组件。 + + 服务账号 uid 格式 ``svc:channel:{channel_type}:{account_id}`` 含冒号， + 用作宿主机路径组件时需转义为 ``_``，与 sandbox_provisioner.app. + LocalContainerProvisionerBackend._validate_uid 保持一致的转义规则， + 确保宿主机路径与容器挂载路径映射一致。 + """ + candidate = str(uid or "").strip() + if not candidate: raise ValueError("uid is required") - if not _SAFE_ID_RE.match(value): - raise ValueError("uid contains invalid characters") - return value + return _UNSAFE_PATH_SEGMENT_CHAR_RE.sub("_", candidate) def _global_user_data_dir(uid: str) -> Path: """Return the shared host-side directory used for one user's workspace files.""" - safe_uid = _validate_uid(uid) + safe_uid = _sanitize_uid(uid) return Path(conf.save_dir) / "threads" / "shared" / safe_uid diff --git a/backend/package/yuxi/channels/adapters/agent_run_adapter.py b/backend/package/yuxi/channels/adapters/agent_run_adapter.py index c6ea2805..a45f33e9 100644 --- a/backend/package/yuxi/channels/adapters/agent_run_adapter.py +++ b/backend/package/yuxi/channels/adapters/agent_run_adapter.py @@ -31,7 +31,11 @@ from sqlalchemy.exc import SQLAlchemyError from yuxi.channels.contract.dtos.messaging.common import MessageContent from yuxi.channels.contract.dtos.plugin.tools import ChannelTool -from yuxi.channels.contract.dtos.shared.agent_run import AgentRunCmd, AgentRunId +from yuxi.channels.contract.dtos.shared.agent_run import ( + AgentRunCmd, + AgentRunId, + AgentRunTerminalInfo, +) from yuxi.channels.contract.dtos.shared.option import Nothing, Option, Some from yuxi.channels.contract.dtos.transport.stream_event import StreamEvent from yuxi.channels.contract.errors import ( @@ -556,3 +560,56 @@ class AgentRunAdapter(AgentRunPort): if text: return Some(text) return Nothing() + + async def getAgentRunTerminalInfo( + self, + run_id: AgentRunId, + ) -> Option[AgentRunTerminalInfo]: + """获取 AgentRun 终态信息。 + + 委托 ``execution_port.getRunTerminalInfo`` 查询 ``agent_runs`` 表当前 + 状态，仅在终态（completed/failed/cancelled/interrupted）时返回 + ``Some(AgentRunTerminalInfo)``，非终态或记录不存在时返回 ``Nothing``。 + + 供 ``LoadBuildStage`` 在事件流为空时根据 AgentRun 终态决定后续处理 + （透传上游错误上下文 / SKIP 投递），避免下游 + ``ValidationError("content", "must not be empty")`` 丢失上下文。 + + 与 ``getAgentRunFinalOutput`` 的区别：本方法读取 ``agent_runs`` 表 + 状态字段（DB），不读取 Redis Stream 内容文本；两者数据源不同、 + 用途不同。 + + Args: + run_id: Agent 运行 ID。 + + Returns: + 终态返回 ``Some(AgentRunTerminalInfo)``；非终态或记录不存在 + 返回 ``Nothing``。 + + Raises: + DependencyError: 依赖服务（DB）故障。 + """ + try: + async with self._session_scope(None) as session: + info = await self._execution_port.getRunTerminalInfo(run_id.value, session) + except (ValidationError, DependencyError, NotFoundError): + raise + except Exception as exc: + await self._logger.error( + "agent_run_terminal_info_failed", + resource="agent_runs", + run_id=run_id.value, + error=str(exc), + ) + raise DependencyError("agent_runs", cause=Error(str(exc))) from exc + + if info is None: + return Nothing() + return Some( + AgentRunTerminalInfo( + status=str(info.get("status") or ""), + agent_id=str(info.get("agent_id") or ""), + error_type=info.get("error_type"), + error_message=info.get("error_message"), + ) + ) diff --git a/backend/package/yuxi/channels/adapters/agent_run_execution_adapter.py b/backend/package/yuxi/channels/adapters/agent_run_execution_adapter.py index d7cc1bd0..91f88534 100644 --- a/backend/package/yuxi/channels/adapters/agent_run_execution_adapter.py +++ b/backend/package/yuxi/channels/adapters/agent_run_execution_adapter.py @@ -21,6 +21,7 @@ from yuxi.channels.contract.ports.driven.shared.agent_run_execution_port import ) from yuxi.services.agent_run_service import ( create_agent_run_view, + get_run_terminal_info, get_run_uid, stream_agent_run_events, ) @@ -137,6 +138,18 @@ class AgentRunExecutionAdapter(AgentRunExecutionPort): """委托 ``agent_run_service.get_run_uid`` 查询运行记录的 uid。""" return await get_run_uid(run_id, db) + async def getRunTerminalInfo( + self, + run_id: str, + db: Any, + ) -> dict[str, Any] | None: + """委托 ``agent_run_service.get_run_terminal_info`` 查询运行记录的终态信息。 + + 返回 ``{status, agent_id, error_type, error_message}`` 字典；非终态 + 或记录不存在时返回 ``None``。 + """ + return await get_run_terminal_info(run_id, db) + def streamAgentRunEvents( self, *, diff --git a/backend/package/yuxi/channels/application/messaging/pipeline/inbound/agent_run_enqueue_stage.py b/backend/package/yuxi/channels/application/messaging/pipeline/inbound/agent_run_enqueue_stage.py index 6a6222eb..cd7e5345 100644 --- a/backend/package/yuxi/channels/application/messaging/pipeline/inbound/agent_run_enqueue_stage.py +++ b/backend/package/yuxi/channels/application/messaging/pipeline/inbound/agent_run_enqueue_stage.py @@ -192,6 +192,10 @@ class AgentRunEnqueueStage: self.logger = logger self.event_publisher = event_publisher self.masking_port = masking_port + # message ops 适配器缺失告警去重（C-N1）：按 channel_type 首次缺失时 + # 记录 WARNING 提示运维，后续同一 channel_type 的入站消息降级为 DEBUG， + # 避免 wechat_woc 等不支持消息操作的渠道在每条消息上都输出 WARNING 噪声。 + self._message_ops_missing_warned: set[ChannelType] = set() async def process(self, context: InboundContext) -> bool: """创建 Agent 运行。 @@ -630,13 +634,27 @@ class AgentRunEnqueueStage: adapter = self.message_ops_adapter_registry.get(context.channel_type) if adapter is None: + # C-N1: 按 channel_type 去重 WARNING 告警。首次缺失时记录 WARNING + # 提示运维（可能是配置遗漏），后续同一 channel_type 的入站消息降级 + # 为 DEBUG（属于"渠道不支持消息操作"的稳定状态，无需重复告警）。 + # 典型场景：wechat_woc 不支持消息操作，注册表无对应 adapter， + # 首条消息 WARNING 后续 DEBUG，避免日志噪声。 if self.logger is not None: - await self.logger.warning( - "message ops adapter not registered, skip message ops enrichment", - trace_id=context.trace_id, - channel_type=context.channel_type, - account_id=context.account_id, - ) + if context.channel_type not in self._message_ops_missing_warned: + self._message_ops_missing_warned.add(context.channel_type) + await self.logger.warning( + "message ops adapter not registered, skip message ops enrichment", + trace_id=context.trace_id, + channel_type=context.channel_type, + account_id=context.account_id, + ) + else: + await self.logger.debug( + "message ops adapter not registered, skip message ops enrichment", + trace_id=context.trace_id, + channel_type=context.channel_type, + account_id=context.account_id, + ) return run_context try: diff --git a/backend/package/yuxi/channels/application/messaging/pipeline/outbound/load_build_stage.py b/backend/package/yuxi/channels/application/messaging/pipeline/outbound/load_build_stage.py index a43d3f53..69da67b2 100644 --- a/backend/package/yuxi/channels/application/messaging/pipeline/outbound/load_build_stage.py +++ b/backend/package/yuxi/channels/application/messaging/pipeline/outbound/load_build_stage.py @@ -15,7 +15,13 @@ from yuxi.channels.application.messaging.context.outbound_context import Outboun from yuxi.channels.contract.dtos.messaging.common import Attachment from yuxi.channels.contract.dtos.messaging.outbound import OutboundPayload, RichMessageFields from yuxi.channels.contract.dtos.shared.agent_run import AgentRunId -from yuxi.channels.contract.dtos.shared.option import Some +from yuxi.channels.contract.dtos.shared.option import Nothing, Some +from yuxi.channels.contract.errors import ( + AgentRunCancelledError, + AgentRunEmptyOutputError, + AgentRunFailedError, + AgentRunInterruptedError, +) from yuxi.channels.contract.plugin.extension_point import FailureStrategy from yuxi.channels.contract.ports.driven.shared.agent_run_port import AgentRunPort @@ -131,9 +137,35 @@ class LoadBuildStage: 已有的 payload 提取逻辑（避免重复解析 StreamEvent payload 嵌套结构）。 + AgentRun 终态感知：当 ``getAgentRunFinalOutput`` 返回 ``Nothing`` + （事件流无 ``messages`` 事件）时，调用 ``getAgentRunTerminalInfo`` + 查询 AgentRun 当前终态，按终态抛出对应的领域错误，透传真实失败 + 原因到 outbound pipeline 调用方： + + - ``failed`` → ``AgentRunFailedError``（携带 run_error_type / + run_error_message，如 ``worker_error: uid contains invalid + characters``），调用方按 delivery failure 处理； + - ``cancelled`` → ``AgentRunCancelledError``，调用方应 SKIP 后续 + 投递（用户主动取消，无需重试）； + - ``interrupted`` → ``AgentRunInterruptedError``，调用方应 SKIP + 后续投递（等待用户输入，有专门通知链路）； + - ``completed`` 但输出为空 → ``AgentRunEmptyOutputError``，表明 + Agent 配置异常或 LLM 返回空内容（如仅产出 ``reasoning_content`` + 或全程只发起 tool_call），需排查 agent 配置与模型行为； + - 非终态（``pending`` / ``running``）或记录不存在 → 返回 + ``Nothing``，本方法不抛错，由下游 ``PrefixStage`` 按原有逻辑 + 抛 ``ValidationError("content", "must not be empty")``。这是 + 极端竞态（DB 状态滞后于事件流）的兜底路径，不影响主流程。 + 参数： context: 出站管道上下文，读取 ``agent_run_id``，写入 ``stream_chunks``。 + + 抛出： + AgentRunFailedError: AgentRun 终态为 ``failed``。 + AgentRunCancelledError: AgentRun 终态为 ``cancelled``。 + AgentRunInterruptedError: AgentRun 终态为 ``interrupted``。 + AgentRunEmptyOutputError: AgentRun 终态为 ``completed`` 但输出为空。 """ run_id = AgentRunId(value=context.agent_run_id) # 步骤 1：阻塞消费至完成（不提取内容，仅等待 end 信号） @@ -144,3 +176,57 @@ class LoadBuildStage: result = await self._agent_run_port.getAgentRunFinalOutput(run_id) if isinstance(result, Some): context.stream_chunks.append(result.value) + return + # 步骤 3：输出为空时查询 AgentRun 终态，按终态抛出领域错误透传上下文 + assert isinstance(result, Nothing) + await self._raiseOnTerminalStatus(context, run_id) + + async def _raiseOnTerminalStatus( + self, + context: OutboundContext, + run_id: AgentRunId, + ) -> None: + """根据 AgentRun 终态抛出对应领域错误。 + + 仅在 ``getAgentRunFinalOutput`` 返回 ``Nothing`` 时调用。查询 + ``agent_runs`` 表当前状态，按终态抛出对应错误。非终态或记录不存在 + 时不抛错（由下游 ``PrefixStage`` 兜底抛 ``ValidationError``）。 + """ + info_option = await self._agent_run_port.getAgentRunTerminalInfo(run_id) + if not isinstance(info_option, Some): + # 非终态或记录不存在：不抛错，下游 PrefixStage 兜底 + return + info = info_option.value + trace_id = context.trace_id + run_id_str = run_id.value + agent_id = info.agent_id + + if info.status == "failed": + raise AgentRunFailedError( + run_id=run_id_str, + agent_id=agent_id, + error_type=info.error_type or "unknown", + error_message=info.error_message or "", + trace_id=trace_id, + ) + if info.status == "cancelled": + raise AgentRunCancelledError( + run_id=run_id_str, + agent_id=agent_id, + trace_id=trace_id, + ) + if info.status == "interrupted": + raise AgentRunInterruptedError( + run_id=run_id_str, + agent_id=agent_id, + error_message=info.error_message or "", + trace_id=trace_id, + ) + if info.status == "completed": + raise AgentRunEmptyOutputError( + run_id=run_id_str, + agent_id=agent_id, + trace_id=trace_id, + ) + # 其他终态值（理论不存在，因 TERMINAL_RUN_STATUSES 仅含 4 种）： + # 不抛错，下游 PrefixStage 兜底 diff --git a/backend/package/yuxi/channels/application/messaging/usecase/inbound_message_service.py b/backend/package/yuxi/channels/application/messaging/usecase/inbound_message_service.py index 7a3c1731..90267beb 100644 --- a/backend/package/yuxi/channels/application/messaging/usecase/inbound_message_service.py +++ b/backend/package/yuxi/channels/application/messaging/usecase/inbound_message_service.py @@ -49,6 +49,8 @@ from yuxi.channels.contract.dtos.messaging.inbound import ( SignatureVerifyResult, ) from yuxi.channels.contract.errors import ( + AgentRunCancelledError, + AgentRunInterruptedError, ConflictError, Error, IdempotencyConflictError, @@ -681,6 +683,28 @@ class InboundMessageService: await self._cache_port.releaseAdvisoryLock(lock_token) if not ok: + # SKIP 语义：AgentRun cancelled / interrupted 由 LoadBuildStage + # 抛出对应领域错误。此类"失败"并非出站投递故障，而是用户主动 + # 取消（cancelled）或 AgentRun 等待用户输入（interrupted，有 + # 专门通知链路），不应触发 NACK / 幂等回退 / 重试，仅记录 INFO + # 日志说明跳过原因并正常结束出站链路。其余错误（含 + # ``AgentRunFailedError`` / ``AgentRunEmptyOutputError``） + # 仍按 delivery failure 处理，透传真实失败原因到错误日志。 + if isinstance(err, (AgentRunCancelledError, AgentRunInterruptedError)): + await self._logger.info( + "outbound pipeline skipped due to agent run terminal state", + trace_id=ctx.trace_id, + agent_run_id=ctx.agent_run_id, + channel_type=ctx.channel_type, + account_id=ctx.account_id, + skip_reason=err.error_code, + run_id=getattr(err, "run_id", ""), + agent_id=getattr(err, "agent_id", ""), + ) + # 跳过 _handleOutboundFailure：不设置 NACK、不回退幂等记录。 + # 出站投递异步化后 ack_decision 保持入站管道决策（ACK 互斥 + # 语义不变），allow transport 层继续接收下一条消息。 + return error_occurred = True err_msg = err.message if err is not None else "pipeline returned failure without error" await self._logger.error( diff --git a/backend/package/yuxi/channels/contract/dtos/shared/agent_run.py b/backend/package/yuxi/channels/contract/dtos/shared/agent_run.py index 504e6a4b..fbaa29f0 100644 --- a/backend/package/yuxi/channels/contract/dtos/shared/agent_run.py +++ b/backend/package/yuxi/channels/contract/dtos/shared/agent_run.py @@ -34,6 +34,32 @@ class AgentRunId: value: str +@dataclass(frozen=True) +class AgentRunTerminalInfo: + """AgentRun 终态信息。 + + 由 ``AgentRunPort.getAgentRunTerminalInfo`` 返回，携带 AgentRun 的最终 + 状态与错误信息，供 ``LoadBuildStage`` 根据 AgentRun 终态决定后续处理 + （正常投递 / 抛出领域错误 / 跳过投递），避免 AgentRun 失败时下游 + 抛出无上下文的 ``ValidationError("content", "must not be empty")``。 + + 字段： + status: AgentRun 终态（``completed`` / ``failed`` / ``cancelled`` + / ``interrupted``）。由 ``agent_runs.status`` 字段读取，仅 + 返回终态值。 + agent_id: Agent slug，用于错误信息上下文（默认空串）。 + error_type: 失败时的错误类型（仅 ``failed`` / ``interrupted`` + 携带，默认 None）。 + error_message: 失败时的错误消息（仅 ``failed`` / ``interrupted`` + 携带，默认 None）。 + """ + + status: str + agent_id: str = "" + error_type: str | None = None + error_message: str | None = None + + @dataclass(frozen=True) class ChannelFormatSpec: """渠道格式规格。 diff --git a/backend/package/yuxi/channels/contract/errors/__init__.py b/backend/package/yuxi/channels/contract/errors/__init__.py index 7ae1f76c..8fcbe7c0 100644 --- a/backend/package/yuxi/channels/contract/errors/__init__.py +++ b/backend/package/yuxi/channels/contract/errors/__init__.py @@ -36,6 +36,10 @@ from yuxi.channels.contract.errors.domain import ( AgentCollaborationError, AgentHandoffFailedError, AgentNotAvailableError, + AgentRunCancelledError, + AgentRunEmptyOutputError, + AgentRunFailedError, + AgentRunInterruptedError, BotLoopBudgetExceededError, CapabilityNotProvenError, ChannelDegradedError, @@ -126,6 +130,11 @@ __all__ = [ "AgentCollaborationError", "AgentNotAvailableError", "AgentHandoffFailedError", + # AgentRun 终态感知相关错误（P0：AgentRun 失败时透传上游错误上下文） + "AgentRunFailedError", + "AgentRunEmptyOutputError", + "AgentRunCancelledError", + "AgentRunInterruptedError", "ContentReviewError", "ContentViolationError", "CredentialNotFoundError", diff --git a/backend/package/yuxi/channels/contract/errors/_status_map.py b/backend/package/yuxi/channels/contract/errors/_status_map.py index a67251ef..c717b65c 100644 --- a/backend/package/yuxi/channels/contract/errors/_status_map.py +++ b/backend/package/yuxi/channels/contract/errors/_status_map.py @@ -61,4 +61,15 @@ CHANNEL_ERROR_STATUS_MAP: dict[str, int] = { "ILLEGAL_STATE": 409, # 媒体下载大小限制相关错误（N-M2） "PAYLOAD_TOO_LARGE": 413, + # AgentRun 终态感知相关错误（P0：AgentRun 失败时透传上游错误上下文） + # - AGENT_RUN_FAILED / AGENT_RUN_EMPTY_OUTPUT：上游 Agent 执行失败或产出 + # 空内容，502 Bad Gateway 语义（上游依赖故障）。 + # - AGENT_RUN_CANCELLED：AgentRun 已被取消（用户主动终止），410 Gone 语义。 + # 经 ``InboundMessageService`` SKIP 后通常不触达 HTTP 层，仅作完整性映射。 + # - AGENT_RUN_INTERRUPTED：AgentRun 被挂起等待用户输入（ask_user_question / + # human_approval），409 Conflict 语义，需要客户端解决状态冲突。 + "AGENT_RUN_FAILED": 502, + "AGENT_RUN_EMPTY_OUTPUT": 502, + "AGENT_RUN_CANCELLED": 410, + "AGENT_RUN_INTERRUPTED": 409, } diff --git a/backend/package/yuxi/channels/contract/errors/domain/__init__.py b/backend/package/yuxi/channels/contract/errors/domain/__init__.py index 4dfbdec4..c3aa381e 100644 --- a/backend/package/yuxi/channels/contract/errors/domain/__init__.py +++ b/backend/package/yuxi/channels/contract/errors/domain/__init__.py @@ -23,6 +23,10 @@ from yuxi.channels.contract.errors.domain.agent import ( AgentCollaborationError, AgentHandoffFailedError, AgentNotAvailableError, + AgentRunCancelledError, + AgentRunEmptyOutputError, + AgentRunFailedError, + AgentRunInterruptedError, ) from yuxi.channels.contract.errors.domain.base import ( BotLoopBudgetExceededError, @@ -101,6 +105,10 @@ __all__ = [ "AgentCollaborationError", "AgentNotAvailableError", "AgentHandoffFailedError", + "AgentRunFailedError", + "AgentRunEmptyOutputError", + "AgentRunCancelledError", + "AgentRunInterruptedError", # content "ContentReviewError", "ContentViolationError", diff --git a/backend/package/yuxi/channels/contract/errors/domain/agent.py b/backend/package/yuxi/channels/contract/errors/domain/agent.py index bf0fbef6..e407b1ee 100644 --- a/backend/package/yuxi/channels/contract/errors/domain/agent.py +++ b/backend/package/yuxi/channels/contract/errors/domain/agent.py @@ -2,10 +2,19 @@ 定义 ``AgentCollaborationError`` 基类及其 2 个子类，覆盖多 Agent 协作场景 中目标 Agent 不可用、移交失败等异常。所有错误继承 ``DomainError``。 + +另定义 ``AgentRunFailedError`` / ``AgentRunEmptyOutputError`` / +``AgentRunCancelledError`` / ``AgentRunInterruptedError`` 四个 AgentRun +终态相关错误，由 ``LoadBuildStage`` 在 AgentRun 进入非正常终态时抛出， +透传真实失败原因（run_id / agent_id / error_type / error_message）到 +outbound pipeline 调用方，避免下游 ``PrefixStage`` 抛出无上下文的 +``ValidationError("content", "must not be empty")``。 """ from __future__ import annotations +from typing import Any + from yuxi.channels.contract.errors.domain.base import DomainError @@ -35,3 +44,146 @@ class AgentHandoffFailedError(AgentCollaborationError): """ error_code = "AGENT_HANDOFF_FAILED" + + +class AgentRunFailedError(DomainError): + """AgentRun 执行失败错误。 + + AgentRun 终态为 ``failed`` 时由 ``LoadBuildStage`` 抛出，携带 + ``run_id`` / ``agent_id`` / ``run_error_type`` / ``run_error_message``， + 透传 Agent 失败的真实原因（如 ``worker_error: uid contains invalid + characters``）到 outbound pipeline 调用方，避免下游 ``PrefixStage`` + 抛出无上下文的 ``ValidationError("content", "must not be empty")``。 + + 调用方（``InboundMessageService._deliverAgentResponse``）应将此错误 + 当作 delivery failure 处理（与原有 ``ok=False`` 路径一致），但错误 + 日志会包含真实失败原因，便于运维排查。 + """ + + error_code = "AGENT_RUN_FAILED" + + def __init__( + self, + *, + run_id: str, + agent_id: str, + error_type: str, + error_message: str, + trace_id: str | None = None, + ) -> None: + super().__init__( + f"AgentRun {run_id} failed (agent={agent_id}, " + f"error_type={error_type}): {error_message}", + trace_id=trace_id, + ) + self.run_id = run_id + self.agent_id = agent_id + self.run_error_type = error_type + self.run_error_message = error_message + + def to_dict(self) -> dict[str, Any]: + data = super().to_dict() + data["run_id"] = self.run_id + data["agent_id"] = self.agent_id + data["run_error_type"] = self.run_error_type + data["run_error_message"] = self.run_error_message + return data + + +class AgentRunEmptyOutputError(DomainError): + """AgentRun 完成但无输出错误。 + + AgentRun 终态为 ``completed`` 但 ``getAgentRunFinalOutput`` 返回 + ``Nothing`` 时由 ``LoadBuildStage`` 抛出，表明 Agent 配置异常或 LLM + 返回空内容（如仅产出 ``reasoning_content`` 或全程只发起 tool_call）， + 需要排查 agent 配置与模型行为。 + """ + + error_code = "AGENT_RUN_EMPTY_OUTPUT" + + def __init__( + self, + *, + run_id: str, + agent_id: str, + trace_id: str | None = None, + ) -> None: + super().__init__( + f"AgentRun {run_id} completed but produced no output (agent={agent_id})", + trace_id=trace_id, + ) + self.run_id = run_id + self.agent_id = agent_id + + def to_dict(self) -> dict[str, Any]: + data = super().to_dict() + data["run_id"] = self.run_id + data["agent_id"] = self.agent_id + return data + + +class AgentRunCancelledError(DomainError): + """AgentRun 被取消错误。 + + AgentRun 终态为 ``cancelled`` 时由 ``LoadBuildStage`` 抛出。调用方 + （``InboundMessageService._deliverAgentResponse``）应捕获此错误并 + 跳过 delivery failure 处理（用户已主动取消，无需再发消息、无需重试）， + 仅记录 INFO 级日志说明跳过原因。 + """ + + error_code = "AGENT_RUN_CANCELLED" + + def __init__( + self, + *, + run_id: str, + agent_id: str, + trace_id: str | None = None, + ) -> None: + super().__init__( + f"AgentRun {run_id} cancelled (agent={agent_id})", + trace_id=trace_id, + ) + self.run_id = run_id + self.agent_id = agent_id + + def to_dict(self) -> dict[str, Any]: + data = super().to_dict() + data["run_id"] = self.run_id + data["agent_id"] = self.agent_id + return data + + +class AgentRunInterruptedError(DomainError): + """AgentRun 被中断错误。 + + AgentRun 终态为 ``interrupted``（需要用户追问 / 人工审批）时由 + ``LoadBuildStage`` 抛出。此类 AgentRun 不应走持久化 outbound 投递 + 路径（追问/审批有专门的通知链路），调用方应捕获此错误并跳过 delivery + failure 处理，仅记录 INFO 级日志说明跳过原因。 + """ + + error_code = "AGENT_RUN_INTERRUPTED" + + def __init__( + self, + *, + run_id: str, + agent_id: str, + error_message: str = "", + trace_id: str | None = None, + ) -> None: + super().__init__( + f"AgentRun {run_id} interrupted (agent={agent_id}): {error_message}", + trace_id=trace_id, + ) + self.run_id = run_id + self.agent_id = agent_id + self.run_error_message = error_message + + def to_dict(self) -> dict[str, Any]: + data = super().to_dict() + data["run_id"] = self.run_id + data["agent_id"] = self.agent_id + data["run_error_message"] = self.run_error_message + return data diff --git a/backend/package/yuxi/channels/contract/policy/config_schema.py b/backend/package/yuxi/channels/contract/policy/config_schema.py index 9448ada1..d121509c 100644 --- a/backend/package/yuxi/channels/contract/policy/config_schema.py +++ b/backend/package/yuxi/channels/contract/policy/config_schema.py @@ -96,7 +96,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=False, hot_reloadable=True, - scope=ConfigScope.CHANNEL, + scope=ConfigScope.GLOBAL, title="Agent 提示开关", description="是否在消息中附加 Agent 提示上下文。", category="功能开关", @@ -107,7 +107,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=False, hot_reloadable=True, - scope=ConfigScope.CHANNEL, + scope=ConfigScope.GLOBAL, title="渠道工具开关", description="是否允许 Agent 调用渠道相关工具。", category="功能开关", @@ -119,7 +119,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=True, hot_reloadable=True, - scope=ConfigScope.CHANNEL, + scope=ConfigScope.GLOBAL, title="消息操作开关", description="总开关：是否启用消息操作（反应、置顶、卡片更新等）。", category="功能开关", @@ -130,7 +130,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=False, hot_reloadable=True, - scope=ConfigScope.CHANNEL, + scope=ConfigScope.ACCOUNT, title="流式响应开关", description="是否启用流式（打字机效果）响应输出。", category="功能开关", @@ -141,7 +141,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=True, hot_reloadable=True, - scope=ConfigScope.CHANNEL, + scope=ConfigScope.ACCOUNT, title="输入中提示开关", description="是否向用户展示'正在输入'状态提示。", category="功能开关", @@ -292,7 +292,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=200, hot_reloadable=True, - scope=ConfigScope.GLOBAL, + scope=ConfigScope.ACCOUNT, title="流式最小间隔", description="流式响应相邻分块之间的最小间隔（毫秒）。", category="流式与传输体验", @@ -303,7 +303,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=60, hot_reloadable=True, - scope=ConfigScope.GLOBAL, + scope=ConfigScope.ACCOUNT, title="流式 TTL", description="流式响应状态的最大保留时间（秒）。", category="流式与传输体验", @@ -314,7 +314,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default=10000, hot_reloadable=True, - scope=ConfigScope.GLOBAL, + scope=ConfigScope.ACCOUNT, title="输入提示 TTL", description="'正在输入'提示状态的最大持续时间（毫秒）。", category="流式与传输体验", @@ -339,7 +339,7 @@ CONFIG_SCHEMA: tuple[ConfigField, ...] = ( required=False, default="after_agent_dispatch", hot_reloadable=True, - scope=ConfigScope.GLOBAL, + scope=ConfigScope.ACCOUNT, title="ACK 策略", description="分阶段 ACK 策略：after_record / after_agent_dispatch / after_persist / manual。", category="可靠性", diff --git a/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_execution_port.py b/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_execution_port.py index d4c48ac3..7c6e37a0 100644 --- a/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_execution_port.py +++ b/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_execution_port.py @@ -8,6 +8,7 @@ DB 技术访问逻辑，供 ``AgentRunAdapter`` 通过本端口调用，消除 ``yuxi.services.run_queue_service`` 中的实际签名保持一致： - ``createAgentRunView`` ← ``agent_run_service.create_agent_run_view`` - ``getRunUid`` ← ``agent_run_service.get_run_uid`` +- ``getRunTerminalInfo`` ← ``agent_run_service.get_run_terminal_info`` - ``streamAgentRunEvents`` ← ``agent_run_service.stream_agent_run_events`` - ``listRunStreamEvents`` ← ``run_queue_service.list_run_stream_events`` @@ -83,6 +84,33 @@ class AgentRunExecutionPort(Protocol): """ ... + async def getRunTerminalInfo( + self, + run_id: str, + db: Any, + ) -> dict[str, Any] | None: + """查询 run 记录的终态信息。 + + 委托 ``agent_run_service.get_run_terminal_info`` 实现，供 + ``AgentRunAdapter.getAgentRunTerminalInfo`` 转换为 + ``AgentRunTerminalInfo`` DTO。仅返回 status / agent_id / error_type / + error_message 四个字段，避免泄露完整 ORM 记录。 + + 非终态（pending/running/cancel_requested）或记录不存在时返回 + ``None``；终态（completed/failed/cancelled/interrupted）返回字典。 + + @pre + - run_id 非空 + + @post + - 终态返回 ``{status, agent_id, error_type, error_message}``； + - 非终态或记录不存在时返回 ``None``。 + + @failure + - DependencyError：DB 故障 + """ + ... + def streamAgentRunEvents( self, *, diff --git a/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_port.py b/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_port.py index d335604a..4bbbeb24 100644 --- a/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_port.py +++ b/backend/package/yuxi/channels/contract/ports/driven/shared/agent_run_port.py @@ -10,7 +10,11 @@ from __future__ import annotations from collections.abc import AsyncIterator from typing import TYPE_CHECKING, Protocol, runtime_checkable -from yuxi.channels.contract.dtos.shared.agent_run import AgentRunCmd, AgentRunId +from yuxi.channels.contract.dtos.shared.agent_run import ( + AgentRunCmd, + AgentRunId, + AgentRunTerminalInfo, +) from yuxi.channels.contract.dtos.shared.option import Option from yuxi.channels.contract.dtos.transport.stream_event import StreamEvent @@ -141,3 +145,43 @@ class AgentRunPort(Protocol): - 读取不修改持久化状态，仅读取 """ ... + + async def getAgentRunTerminalInfo(self, run_id: AgentRunId) -> Option[AgentRunTerminalInfo]: + """获取 AgentRun 终态信息。 + + 查询 ``agent_runs`` 表当前状态，返回终态信息（status / agent_id / + error_type / error_message），供 ``LoadBuildStage`` 在事件流为空时 + 根据 AgentRun 终态决定后续处理： + + - ``completed`` + 空输出 → 抛 ``AgentRunEmptyOutputError``，透传上游 + 产出缺陷（避免下游 ``ValidationError("content", "must not be empty")`` + 丢失上下文）； + - ``failed`` → 抛 ``AgentRunFailedError``，透传上游执行错误 + （error_type / error_message）； + - ``cancelled`` → 抛 ``AgentRunCancelledError``，调用方应 SKIP 后续 + 投递（用户主动终止，非投递失败）； + - ``interrupted`` → 抛 ``AgentRunInterruptedError``，调用方应 SKIP + 后续投递（等待用户输入，非投递失败）； + - ``pending`` / ``running`` → 返回 ``Nothing``（AgentRun 未终态， + 由调用方按既有逻辑处理）。 + + 与 ``getAgentRunFinalOutput`` 的区别：本方法读取 ``agent_runs`` 表 + 状态字段，不读取 Redis Stream 事件流；``getAgentRunFinalOutput`` + 读取 Redis Stream 内容文本。两者数据源不同、用途不同。 + + @pre + - run_id 非空且为已创建的 Agent 运行 ID + + @post + - 终态（completed/failed/cancelled/interrupted）返回 + ``Some(AgentRunTerminalInfo)``； + - 非终态（pending/running）或记录不存在时返回 ``Nothing``。 + + @failure + - DependencyError：依赖服务（DB）故障 + + @consistency + - 最终一致性：状态字段可能滞后于 Agent 实际执行状态 + - 读取不修改持久化状态，仅读取 + """ + ... diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py index ba71d3bf..a1c9bd28 100644 --- a/backend/package/yuxi/services/agent_run_service.py +++ b/backend/package/yuxi/services/agent_run_service.py @@ -7,6 +7,7 @@ import json import os import uuid from collections.abc import AsyncIterator +from typing import Any from fastapi import HTTPException from sqlalchemy import select @@ -398,6 +399,32 @@ async def get_run_uid(run_id: str, db: AsyncSession) -> str: return str(run.uid) +async def get_run_terminal_info(run_id: str, db: AsyncSession) -> dict[str, Any] | None: + """查询 run 记录的终态信息。 + + 仅返回 status / agent_id / error_type / error_message 四个字段。run 不存在 + 或非终态（pending/running/cancel_requested）时返回 ``None``；终态 + （completed/failed/cancelled/interrupted）返回字典。 + + 供 ``AgentRunAdapter.getAgentRunTerminalInfo`` 转换为 + ``AgentRunTerminalInfo`` DTO，使 ``LoadBuildStage`` 能在事件流为空时 + 根据 AgentRun 终态决定后续处理（透传上游错误上下文 / SKIP 投递）， + 而不是抛出无上下文的 ``ValidationError("content", "must not be empty")``。 + """ + repo = AgentRunRepository(db) + run = await repo.get_run(run_id) + if run is None: + return None + if run.status not in TERMINAL_RUN_STATUSES: + return None + return { + "status": run.status, + "agent_id": run.agent_id, + "error_type": run.error_type, + "error_message": run.error_message, + } + + async def cancel_agent_run_view(*, run_id: str, current_uid: str, db: AsyncSession) -> dict: repo = AgentRunRepository(db) run = await repo.get_run_for_user(run_id, str(current_uid)) diff --git a/backend/test/unit/backends/test_sandbox_paths.py b/backend/test/unit/backends/test_sandbox_paths.py new file mode 100644 index 00000000..6fca9938 --- /dev/null +++ b/backend/test/unit/backends/test_sandbox_paths.py @@ -0,0 +1,44 @@ +"""Tests for yuxi.agents.backends.sandbox.paths uid sanitize behavior. + +服务账号 uid 格式 ``svc:channel:{channel_type}:{account_id}`` 含冒号， +paths.py._sanitize_uid 必须将其转义为文件系统安全的路径组件，且与 +docker/sandbox_provisioner/app.py 中的转义规则保持一致，确保宿主机 +路径与容器挂载路径映射一致。 +""" + +from __future__ import annotations + +import pytest + +from yuxi.agents.backends.sandbox.paths import _sanitize_uid + + +def test_sanitize_uid_passes_through_safe_chars() -> None: + assert _sanitize_uid("user-1_2") == "user-1_2" + assert _sanitize_uid("abcDEF123") == "abcDEF123" + + +def test_sanitize_uid_escapes_colons_in_service_account_uid() -> None: + # 服务账号 uid: svc:channel:{channel_type}:{account_id} + assert _sanitize_uid("svc:channel:wechat_woc:acc-1") == "svc_channel_wechat_woc_acc-1" + assert _sanitize_uid("svc:channel:feishu:host_1_bridge_2") == "svc_channel_feishu_host_1_bridge_2" + + +def test_sanitize_uid_escapes_unsafe_chars_to_underscore() -> None: + assert _sanitize_uid("user/name") == "user_name" + assert _sanitize_uid("user name") == "user_name" + assert _sanitize_uid("user;rm") == "user_rm" + assert _sanitize_uid("user.name") == "user_name" + # 路径穿越尝试被转义后不再包含 ..（每个非法字符独立替换为 _） + assert _sanitize_uid("../user") == "___user" + assert _sanitize_uid("..") == "__" + + +def test_sanitize_uid_rejects_empty_or_whitespace() -> None: + for value in ["", " ", None]: + with pytest.raises(ValueError): + _sanitize_uid(value) # type: ignore[arg-type] + + +def test_sanitize_uid_strips_surrounding_whitespace() -> None: + assert _sanitize_uid(" user-1 ") == "user-1" diff --git a/backend/test/unit/backends/test_sandbox_provisioner_config.py b/backend/test/unit/backends/test_sandbox_provisioner_config.py index 612f7e81..c388140f 100644 --- a/backend/test/unit/backends/test_sandbox_provisioner_config.py +++ b/backend/test/unit/backends/test_sandbox_provisioner_config.py @@ -78,7 +78,16 @@ def test_local_container_identity_validation_rejects_unsafe_path_segments(monkey with pytest.raises(ValueError): backend_cls._validate_thread_id(value) - for value in ["../user", "user/name", "user name", "user;rm", "user.name"]: + # thread_id 仍为 strict 校验（拒绝非法字符）；uid 改为 sanitize 风格， + # 含冒号的服务账号 uid（svc:channel:{type}:{account_id}）转义为 _ 后可用。 + assert backend_cls._validate_uid("svc:channel:wechat_woc:acc-1") == "svc_channel_wechat_woc_acc-1" + assert backend_cls._validate_uid("../user") == "___user" + assert backend_cls._validate_uid("user/name") == "user_name" + assert backend_cls._validate_uid("user name") == "user_name" + assert backend_cls._validate_uid("user;rm") == "user_rm" + assert backend_cls._validate_uid("user.name") == "user_name" + + for value in ["", " ", None]: with pytest.raises(ValueError): backend_cls._validate_uid(value) diff --git a/backend/test/unit/channels/adapters/test_agent_run_adapter.py b/backend/test/unit/channels/adapters/test_agent_run_adapter.py index d2d44d7a..bc05cc8b 100644 --- a/backend/test/unit/channels/adapters/test_agent_run_adapter.py +++ b/backend/test/unit/channels/adapters/test_agent_run_adapter.py @@ -1,10 +1,10 @@ """yuxi.channels.adapters.agent_run_adapter 单元测试。 覆盖 ``AgentRunAdapter`` 的 ``createAgentRun`` / ``streamAgentRun`` / -``getAgentRunFinalOutput`` 方法，使用 ``AsyncMock`` 注入 -``AgentRunExecutionPort`` 与 ``ServiceAccountPort`` 桩，patch -``pg_manager.get_async_session_context`` 返回 mock session，不连接真实 -Redis / DB。 +``getAgentRunFinalOutput`` / ``getAgentRunTerminalInfo`` 方法，使用 +``AsyncMock`` 注入 ``AgentRunExecutionPort`` 与 ``ServiceAccountPort`` 桩， +patch ``pg_manager.get_async_session_context`` 返回 mock session，不连接 +真实 Redis / DB。 createAgentRun 通过 ServiceAccountPort 解析执行身份（服务账号 uid）， 并通过 pg_manager 获取真实 AsyncSession 注入下游；测试覆盖以下路径： @@ -13,6 +13,13 @@ createAgentRun 通过 ServiceAccountPort 解析执行身份（服务账号 uid - 渠道来源完整：解析服务账号，current_uid 使用服务账号 uid - 服务账号缺失：抛 InternalError（INV-7 错误显式化，不回退） - 服务账号异常：契约错误放行，其余包装为 DependencyError（ADP-002） + +getAgentRunTerminalInfo 委托 ``execution_port.getRunTerminalInfo`` 查询 +``agent_runs`` 表终态信息；测试覆盖以下路径： +- 终态返回 Some(AgentRunTerminalInfo)，含 status / agent_id / error_type / error_message +- 非终态或记录不存在返回 Nothing +- 契约错误（ValidationError / DependencyError / NotFoundError）放行 +- 其余异常包装为 DependencyError（ADP-002） """ from __future__ import annotations @@ -28,12 +35,14 @@ from yuxi.channels.contract.dtos.shared.agent_run import ( AgentRunCmd, AgentRunContext, AgentRunId, + AgentRunTerminalInfo, ) from yuxi.channels.contract.dtos.shared.option import Nothing, Some from yuxi.channels.contract.dtos.shared.service_account import ServiceAccount from yuxi.channels.contract.dtos.transport.stream_event import StreamEvent from yuxi.channels.contract.errors import ( DependencyError, + Error, InternalError, NotFoundError, ValidationError, @@ -739,3 +748,262 @@ class TestAgentRunAdapterGetFinalOutput: assert isinstance(result, Some) assert result.value == "x" * 200 assert execution_port.listRunStreamEvents.await_count == 2 + + +@pytest.mark.unit +class TestAgentRunAdapterGetTerminalInfo: + """``getAgentRunTerminalInfo`` 行为测试。 + + 委托 ``execution_port.getRunTerminalInfo`` 查询 ``agent_runs`` 表当前 + 状态。终态（completed/failed/cancelled/interrupted）返回 + ``Some(AgentRunTerminalInfo)``，非终态或记录不存在返回 ``Nothing``。 + + 注意：``getAgentRunTerminalInfo`` 通过 ``_session_scope(None)`` 调用 + ``pg_manager.get_async_session_context``，测试需 patch pg_manager + 避免连接真实 DB。 + """ + + @pytest.mark.asyncio + async def test_returns_some_when_run_in_terminal_status_failed(self): + # Arrange — AgentRun failed 终态，含全部错误字段 + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + return_value={ + "status": "failed", + "agent_id": "default-chatbot", + "error_type": "worker_error", + "error_message": "uid contains invalid characters", + } + ) + logger = MagicMock() + logger.error = AsyncMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-failed-1")) + + # Assert + assert isinstance(result, Some) + info = result.value + assert isinstance(info, AgentRunTerminalInfo) + assert info.status == "failed" + assert info.agent_id == "default-chatbot" + assert info.error_type == "worker_error" + assert info.error_message == "uid contains invalid characters" + + @pytest.mark.asyncio + async def test_returns_some_when_run_in_terminal_status_cancelled(self): + # Arrange — AgentRun cancelled 终态，无 error_type / error_message + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + return_value={ + "status": "cancelled", + "agent_id": "default-chatbot", + "error_type": None, + "error_message": None, + } + ) + logger = MagicMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-cancelled-1")) + + # Assert + assert isinstance(result, Some) + info = result.value + assert info.status == "cancelled" + assert info.agent_id == "default-chatbot" + assert info.error_type is None + assert info.error_message is None + + @pytest.mark.asyncio + async def test_returns_some_when_run_in_terminal_status_interrupted(self): + # Arrange — AgentRun interrupted 终态 + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + return_value={ + "status": "interrupted", + "agent_id": "default-chatbot", + "error_type": None, + "error_message": "waiting for user input", + } + ) + logger = MagicMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-interrupted-1")) + + # Assert + assert isinstance(result, Some) + info = result.value + assert info.status == "interrupted" + assert info.error_message == "waiting for user input" + + @pytest.mark.asyncio + async def test_returns_some_when_run_in_terminal_status_completed(self): + # Arrange — AgentRun completed 终态 + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + return_value={ + "status": "completed", + "agent_id": "default-chatbot", + "error_type": None, + "error_message": None, + } + ) + logger = MagicMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-completed-1")) + + # Assert + assert isinstance(result, Some) + info = result.value + assert info.status == "completed" + + @pytest.mark.asyncio + async def test_returns_nothing_when_run_not_in_terminal_status(self): + # Arrange — AgentRun 仍为 running / pending（非终态） + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock(return_value=None) + logger = MagicMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-running-1")) + + # Assert + assert isinstance(result, Nothing) + + @pytest.mark.asyncio + async def test_returns_nothing_when_run_record_not_found(self): + # Arrange — AgentRun 记录不存在（已删除或 ID 错误） + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock(return_value=None) + logger = MagicMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-not-exist")) + + # Assert + assert isinstance(result, Nothing) + + @pytest.mark.asyncio + async def test_returns_some_with_empty_strings_when_db_fields_missing(self): + # Arrange — DB 返回的 status / agent_id 为 None（理论不应发生，但 + # adapter 应兜底为空串而非 None，保证 AgentRunTerminalInfo 字段类型） + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + return_value={ + "status": None, + "agent_id": None, + "error_type": None, + "error_message": None, + } + ) + logger = MagicMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act + with _patch_pg_session(): + result = await adapter.getAgentRunTerminalInfo(AgentRunId("run-empty-fields")) + + # Assert — status / agent_id 兜底为空串，error_type / error_message 保持 None + assert isinstance(result, Some) + info = result.value + assert info.status == "" + assert info.agent_id == "" + assert info.error_type is None + assert info.error_message is None + + @pytest.mark.asyncio + async def test_passes_through_validation_error(self): + # Arrange — 契约错误（ValidationError）放行，不包装为 DependencyError + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + side_effect=ValidationError("run_id", "invalid format") + ) + logger = MagicMock() + logger.error = AsyncMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act / Assert + with _patch_pg_session(), pytest.raises(ValidationError): + await adapter.getAgentRunTerminalInfo(AgentRunId("run-1")) + # 不应记录 error 日志（契约错误直接放行） + logger.error.assert_not_awaited() + + @pytest.mark.asyncio + async def test_passes_through_dependency_error(self): + # Arrange — 契约错误（DependencyError）放行 + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + side_effect=DependencyError("agent_runs", cause=Error("db down")) + ) + logger = MagicMock() + logger.error = AsyncMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act / Assert + with _patch_pg_session(), pytest.raises(DependencyError): + await adapter.getAgentRunTerminalInfo(AgentRunId("run-1")) + logger.error.assert_not_awaited() + + @pytest.mark.asyncio + async def test_passes_through_not_found_error(self): + # Arrange — 契约错误（NotFoundError）放行 + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock( + side_effect=NotFoundError("agent_runs", "run not found") + ) + logger = MagicMock() + logger.error = AsyncMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act / Assert + with _patch_pg_session(), pytest.raises(NotFoundError): + await adapter.getAgentRunTerminalInfo(AgentRunId("run-1")) + logger.error.assert_not_awaited() + + @pytest.mark.asyncio + async def test_translates_generic_exception_to_dependency_error(self): + # Arrange — 非契约错误（如 RuntimeError）包装为 DependencyError + # （ADP-002），保留原始异常链（from exc） + execution_port = MagicMock() + execution_port.getRunTerminalInfo = AsyncMock(side_effect=RuntimeError("db connection lost")) + logger = MagicMock() + logger.error = AsyncMock() + service_account_port = MagicMock() + adapter = AgentRunAdapter(execution_port, logger, service_account_port) + + # Act / Assert + with _patch_pg_session(), pytest.raises(DependencyError) as exc_info: + await adapter.getAgentRunTerminalInfo(AgentRunId("run-1")) + # 原始异常链保留 + assert isinstance(exc_info.value.__cause__, RuntimeError) + assert "db connection lost" in str(exc_info.value.__cause__) + # 记录 error 日志（INV-10 可观测性） + logger.error.assert_awaited_once() + log_kwargs = logger.error.call_args.kwargs + assert log_kwargs["resource"] == "agent_runs" + assert log_kwargs["run_id"] == "run-1" diff --git a/backend/test/unit/channels/application/pipeline/outbound/test_load_build_stage.py b/backend/test/unit/channels/application/pipeline/outbound/test_load_build_stage.py index 2682beb9..49992ab6 100644 --- a/backend/test/unit/channels/application/pipeline/outbound/test_load_build_stage.py +++ b/backend/test/unit/channels/application/pipeline/outbound/test_load_build_stage.py @@ -4,6 +4,8 @@ - ``process(context)``：正常路径（含富消息/附件）、无富消息、空分块、 image_url/video_url 转换为附件 - 持久化模式下 AgentRun 内容加载（streamAgentRun 阻塞消费 + getAgentRunFinalOutput） +- AgentRun 终态感知：``failed`` / ``cancelled`` / ``interrupted`` / ``completed`` / + 非终态场景下 ``_loadAgentRunOutput`` 的行为契约 """ from __future__ import annotations @@ -17,8 +19,17 @@ from yuxi.channels.application.messaging.pipeline.outbound.load_build_stage impo from yuxi.channels.contract.dtos.messaging.channel import ChannelType from yuxi.channels.contract.dtos.messaging.common import Attachment from yuxi.channels.contract.dtos.messaging.outbound import RichMessage -from yuxi.channels.contract.dtos.shared.agent_run import AgentRunId +from yuxi.channels.contract.dtos.shared.agent_run import ( + AgentRunId, + AgentRunTerminalInfo, +) from yuxi.channels.contract.dtos.shared.option import Nothing, Some +from yuxi.channels.contract.errors import ( + AgentRunCancelledError, + AgentRunEmptyOutputError, + AgentRunFailedError, + AgentRunInterruptedError, +) from yuxi.channels.contract.plugin.extension_point import FailureStrategy pytestmark = pytest.mark.unit @@ -42,12 +53,15 @@ def _make_ctx(**overrides) -> OutboundContext: def _make_agent_run_port( *, stream_events: list | None = None, - final_output: Some | Nothing = None, + final_output: Some | Nothing | None = None, + terminal_info: Some | Nothing | None = None, ) -> MagicMock: """构造 mock AgentRunPort。 stream_events：streamAgentRun 产出的事件列表（按序 yield 后结束迭代）。 - final_output：getAgentRunFinalOutput 返回的 Option。 + final_output：getAgentRunFinalOutput 返回的 Option，默认 Nothing()。 + terminal_info：getAgentRunTerminalInfo 返回的 Option，默认 Nothing() + （非终态或记录不存在，由下游 PrefixStage 兜底）。 """ port = MagicMock() @@ -57,6 +71,7 @@ def _make_agent_run_port( port.streamAgentRun = MagicMock(side_effect=_stream) port.getAgentRunFinalOutput = AsyncMock(return_value=final_output or Nothing()) + port.getAgentRunTerminalInfo = AsyncMock(return_value=terminal_info or Nothing()) return port @@ -333,11 +348,16 @@ class TestLoadBuildStagePersistentContentLoad: assert ctx.outbound_payload.stream_chunks == ("full response text",) @pytest.mark.asyncio - async def test_persistent_mode_final_output_nothing_keeps_empty(self): - # Arrange — getAgentRunFinalOutput 返回 Nothing（AgentRun 无文本输出） + async def test_persistent_mode_final_output_nothing_non_terminal_keeps_empty(self): + # Arrange — getAgentRunFinalOutput 返回 Nothing（无 messages 事件）， + # 同时 getAgentRunTerminalInfo 返回 Nothing（AgentRun 仍为 pending / + # running，或记录已删除）。此为极端竞态：事件流已结束但 DB 状态 + # 滞后未进入终态。LoadBuildStage 不抛错，交由下游 PrefixStage 兜底 + # 抛 ValidationError("content", "must not be empty")。 port = _make_agent_run_port( stream_events=["evt-1"], final_output=Nothing(), + terminal_info=Nothing(), ) stage = LoadBuildStage(agent_run_port=port) ctx = _make_ctx(delivery_mode="persistent", agent_run_id="run-001", stream_chunks=[]) @@ -345,10 +365,13 @@ class TestLoadBuildStagePersistentContentLoad: # Act ok = await stage.process(ctx) - # Assert — stream_chunks 保持空，不抛异常 + # Assert — stream_chunks 保持空，process 返回 True（终态查询返回 + # Nothing，不抛领域错误；下游 PrefixStage 才会抛 ValidationError） assert ok is True assert ctx.stream_chunks == [] assert ctx.outbound_payload.stream_chunks == () + # getAgentRunTerminalInfo 被调用一次 + port.getAgentRunTerminalInfo.assert_called_once() @pytest.mark.asyncio async def test_streaming_mode_skips_agent_run_load(self): @@ -399,3 +422,270 @@ class TestLoadBuildStagePersistentContentLoad: assert ok is True port.streamAgentRun.assert_not_called() port.getAgentRunFinalOutput.assert_not_called() + + +# ─── AgentRun 终态感知 ────────────────────────────────────────────────────── + + +@pytest.mark.unit +class TestLoadBuildStageTerminalStatusAwareness: + """AgentRun 终态感知：``getAgentRunFinalOutput`` 返回 ``Nothing`` 时按 + ``AgentRunTerminalInfo.status`` 抛出对应领域错误，透传真实失败原因到 + outbound pipeline 调用方。 + + 覆盖 5 个分支： + - ``failed`` → ``AgentRunFailedError``（携带 run_error_type / + run_error_message） + - ``cancelled`` → ``AgentRunCancelledError``（SKIP 语义） + - ``interrupted`` → ``AgentRunInterruptedError``（SKIP 语义） + - ``completed`` 但输出为空 → ``AgentRunEmptyOutputError`` + - 非终态或记录不存在 → 不抛错，下游 PrefixStage 兜底 + """ + + @pytest.mark.asyncio + async def test_failed_status_raises_agent_run_failed_error(self): + # Arrange — AgentRun 终态为 failed，应透传 error_type / error_message + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="failed", + agent_id="default-chatbot", + error_type="worker_error", + error_message="uid contains invalid characters", + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-failed-001", + stream_chunks=[], + ) + + # Act / Assert + with pytest.raises(AgentRunFailedError) as exc_info: + await stage.process(ctx) + + err = exc_info.value + assert err.run_id == "run-failed-001" + assert err.agent_id == "default-chatbot" + assert err.run_error_type == "worker_error" + assert err.run_error_message == "uid contains invalid characters" + assert err.error_code == "AGENT_RUN_FAILED" + assert err.trace_id == "trace-001" + # 错误信息包含真实失败原因，便于运维排查 + assert "worker_error" in err.message + assert "uid contains invalid characters" in err.message + + @pytest.mark.asyncio + async def test_cancelled_status_raises_agent_run_cancelled_error(self): + # Arrange — AgentRun 被取消（用户主动终止），调用方应 SKIP 后续投递 + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="cancelled", + agent_id="default-chatbot", + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-cancelled-001", + stream_chunks=[], + ) + + # Act / Assert + with pytest.raises(AgentRunCancelledError) as exc_info: + await stage.process(ctx) + + err = exc_info.value + assert err.run_id == "run-cancelled-001" + assert err.agent_id == "default-chatbot" + assert err.error_code == "AGENT_RUN_CANCELLED" + assert err.trace_id == "trace-001" + + @pytest.mark.asyncio + async def test_interrupted_status_raises_agent_run_interrupted_error(self): + # Arrange — AgentRun 被中断（等待用户输入 / 人工审批），调用方应 + # SKIP 后续投递（有专门通知链路） + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="interrupted", + agent_id="default-chatbot", + error_message="waiting for user input", + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-interrupted-001", + stream_chunks=[], + ) + + # Act / Assert + with pytest.raises(AgentRunInterruptedError) as exc_info: + await stage.process(ctx) + + err = exc_info.value + assert err.run_id == "run-interrupted-001" + assert err.agent_id == "default-chatbot" + assert err.run_error_message == "waiting for user input" + assert err.error_code == "AGENT_RUN_INTERRUPTED" + assert err.trace_id == "trace-001" + + @pytest.mark.asyncio + async def test_completed_with_empty_output_raises_empty_output_error(self): + # Arrange — AgentRun 终态为 completed 但输出为空（Agent 配置异常或 + # LLM 仅产出 reasoning_content / 全程只发起 tool_call），需排查 + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="completed", + agent_id="default-chatbot", + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-empty-001", + stream_chunks=[], + ) + + # Act / Assert + with pytest.raises(AgentRunEmptyOutputError) as exc_info: + await stage.process(ctx) + + err = exc_info.value + assert err.run_id == "run-empty-001" + assert err.agent_id == "default-chatbot" + assert err.error_code == "AGENT_RUN_EMPTY_OUTPUT" + assert err.trace_id == "trace-001" + + @pytest.mark.asyncio + async def test_failed_status_with_none_error_fields_uses_defaults(self): + # Arrange — AgentRun failed 但 error_type / error_message 为 None + # （历史数据或异常路径未填充），应使用 "unknown" / "" 兜底 + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="failed", + agent_id="default-chatbot", + error_type=None, + error_message=None, + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-failed-002", + stream_chunks=[], + ) + + # Act / Assert + with pytest.raises(AgentRunFailedError) as exc_info: + await stage.process(ctx) + + err = exc_info.value + assert err.run_error_type == "unknown" + assert err.run_error_message == "" + + @pytest.mark.asyncio + async def test_interrupted_status_with_none_error_message_uses_empty(self): + # Arrange — AgentRun interrupted 但 error_message 为 None，应使用空串 + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="interrupted", + agent_id="default-chatbot", + error_message=None, + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-interrupted-002", + stream_chunks=[], + ) + + # Act / Assert + with pytest.raises(AgentRunInterruptedError) as exc_info: + await stage.process(ctx) + + err = exc_info.value + assert err.run_error_message == "" + + @pytest.mark.asyncio + async def test_unknown_terminal_status_does_not_raise(self): + # Arrange — 理论不存在的终态值（TERMINAL_RUN_STATUSES 仅含 4 种）， + # 兜底分支：不抛错，由下游 PrefixStage 处理。验证防御性兜底行为。 + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Nothing(), + terminal_info=Some( + AgentRunTerminalInfo( + status="unknown_status", + agent_id="default-chatbot", + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-unknown-001", + stream_chunks=[], + ) + + # Act + ok = await stage.process(ctx) + + # Assert — 不抛错，process 返回 True（下游 PrefixStage 兜底） + assert ok is True + assert ctx.stream_chunks == [] + + @pytest.mark.asyncio + async def test_terminal_info_not_called_when_final_output_present(self): + # Arrange — getAgentRunFinalOutput 返回 Some（有内容）时不应调用 + # getAgentRunTerminalInfo（短路返回，性能优化） + port = _make_agent_run_port( + stream_events=["evt-1"], + final_output=Some("response text"), + terminal_info=Some( + AgentRunTerminalInfo( + status="completed", + agent_id="default-chatbot", + ), + ), + ) + stage = LoadBuildStage(agent_run_port=port) + ctx = _make_ctx( + delivery_mode="persistent", + agent_run_id="run-ok-001", + stream_chunks=[], + ) + + # Act + ok = await stage.process(ctx) + + # Assert + assert ok is True + port.getAgentRunFinalOutput.assert_called_once() + # 终态查询不应被调用（短路返回） + port.getAgentRunTerminalInfo.assert_not_called() diff --git a/backend/test/unit/channels/application/usecase/test_inbound_message_service.py b/backend/test/unit/channels/application/usecase/test_inbound_message_service.py index e8bd029f..a031c668 100644 --- a/backend/test/unit/channels/application/usecase/test_inbound_message_service.py +++ b/backend/test/unit/channels/application/usecase/test_inbound_message_service.py @@ -33,6 +33,10 @@ from yuxi.channels.contract.dtos.messaging.inbound import ( ) from yuxi.channels.contract.dtos.shared.command import CommandResponse, CommandResult from yuxi.channels.contract.errors import ( + AgentRunCancelledError, + AgentRunEmptyOutputError, + AgentRunFailedError, + AgentRunInterruptedError, DependencyError, Error, IdempotencyConflictError, @@ -853,6 +857,258 @@ class TestOutboundFailureIdempotencyRollback: assert result.ack_decision == "pending" +@pytest.mark.unit +class TestOutboundSkipOnAgentRunTerminalState: + """SKIP 语义：AgentRun cancelled / interrupted 时跳过 delivery failure 处理。 + + 覆盖 ``InboundMessageService._deliverAgentResponse`` 在出站管道返回 + ``ok=False`` 且错误为 ``AgentRunCancelledError`` / + ``AgentRunInterruptedError`` 时，**不**触发 NACK / 幂等回退 / 重试， + 仅记录 INFO 日志说明跳过原因。 + + 与 ``AgentRunFailedError`` / ``AgentRunEmptyOutputError`` 对比：后者仍 + 按 delivery failure 处理（透传真实失败原因到错误日志）。 + """ + + @pytest.mark.asyncio + async def test_cancelled_agent_run_skips_outbound_failure_handling( + self, + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + fake_idempotency_repo, + ): + # Arrange AgentRun cancelled → 出站返回 ok=False 携带 + # AgentRunCancelledError，应跳过 _handleOutboundFailure（不回退幂等） + inbound_pipeline = AsyncMock() + + async def _set_agent_run_with_idempotency(ctx): + ctx.agent_run_id = "run-cancelled-1" + ctx.idempotency_record_id = 42 + return (True, None) + + inbound_pipeline.run.side_effect = _set_agent_run_with_idempotency + outbound_pipeline = AsyncMock() + err = AgentRunCancelledError( + run_id="run-cancelled-1", + agent_id="default-chatbot", + trace_id="trace-1", + ) + outbound_pipeline.run.return_value = (False, err) + streaming_config = MagicMock() + streaming_config.streaming_enabled = False + fake_config.getStreamingConfig.return_value = streaming_config + service = _make_service( + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + inbound_pipeline=inbound_pipeline, + outbound_pipeline=outbound_pipeline, + idempotency_repository=fake_idempotency_repo, + ) + + # Act + result = await service.receiveInbound(_make_inbound_cmd()) + await _drain_outbound_tasks(service) + + # Assert 幂等记录不被回退为 failed（SKIP 语义） + statuses = [call.args[1] for call in fake_idempotency_repo.updateIdempotencyStatus.call_args_list] + assert "failed" not in statuses + # INFO 日志记录跳过原因 + fake_logger.info.assert_awaited() + skip_logs = [ + call for call in fake_logger.info.call_args_list + if "outbound pipeline skipped" in str(call) + ] + assert skip_logs, "应记录 'outbound pipeline skipped' INFO 日志" + # 日志中包含 skip_reason / run_id / agent_id + skip_call_kwargs = skip_logs[0].kwargs + assert skip_call_kwargs["skip_reason"] == "AGENT_RUN_CANCELLED" + assert skip_call_kwargs["run_id"] == "run-cancelled-1" + assert skip_call_kwargs["agent_id"] == "default-chatbot" + # 出站异步化后 ack_decision 保持 pending，无 error 透传 + assert result.ack_decision == "pending" + assert result.error is None + + @pytest.mark.asyncio + async def test_interrupted_agent_run_skips_outbound_failure_handling( + self, + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + fake_idempotency_repo, + ): + # Arrange AgentRun interrupted（等待用户输入）→ 出站返回 ok=False + # 携带 AgentRunInterruptedError，应跳过 _handleOutboundFailure + inbound_pipeline = AsyncMock() + + async def _set_agent_run_with_idempotency(ctx): + ctx.agent_run_id = "run-interrupted-1" + ctx.idempotency_record_id = 42 + return (True, None) + + inbound_pipeline.run.side_effect = _set_agent_run_with_idempotency + outbound_pipeline = AsyncMock() + err = AgentRunInterruptedError( + run_id="run-interrupted-1", + agent_id="default-chatbot", + error_message="waiting for user input", + trace_id="trace-1", + ) + outbound_pipeline.run.return_value = (False, err) + streaming_config = MagicMock() + streaming_config.streaming_enabled = False + fake_config.getStreamingConfig.return_value = streaming_config + service = _make_service( + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + inbound_pipeline=inbound_pipeline, + outbound_pipeline=outbound_pipeline, + idempotency_repository=fake_idempotency_repo, + ) + + # Act + result = await service.receiveInbound(_make_inbound_cmd()) + await _drain_outbound_tasks(service) + + # Assert 幂等记录不被回退为 failed + statuses = [call.args[1] for call in fake_idempotency_repo.updateIdempotencyStatus.call_args_list] + assert "failed" not in statuses + # INFO 日志记录跳过原因 + skip_logs = [ + call for call in fake_logger.info.call_args_list + if "outbound pipeline skipped" in str(call) + ] + assert skip_logs + skip_call_kwargs = skip_logs[0].kwargs + assert skip_call_kwargs["skip_reason"] == "AGENT_RUN_INTERRUPTED" + assert skip_call_kwargs["run_id"] == "run-interrupted-1" + assert result.ack_decision == "pending" + assert result.error is None + + @pytest.mark.asyncio + async def test_failed_agent_run_still_triggers_outbound_failure_handling( + self, + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + fake_idempotency_repo, + ): + # Arrange AgentRun failed（非 SKIP 类错误）→ 出站返回 ok=False + # 携带 AgentRunFailedError，应仍按 delivery failure 处理（回退幂等 + # 为 failed，允许渠道重试），透传真实失败原因到错误日志 + inbound_pipeline = AsyncMock() + + async def _set_agent_run_with_idempotency(ctx): + ctx.agent_run_id = "run-failed-1" + ctx.idempotency_record_id = 42 + return (True, None) + + inbound_pipeline.run.side_effect = _set_agent_run_with_idempotency + outbound_pipeline = AsyncMock() + err = AgentRunFailedError( + run_id="run-failed-1", + agent_id="default-chatbot", + error_type="worker_error", + error_message="uid contains invalid characters", + trace_id="trace-1", + ) + outbound_pipeline.run.return_value = (False, err) + streaming_config = MagicMock() + streaming_config.streaming_enabled = False + fake_config.getStreamingConfig.return_value = streaming_config + service = _make_service( + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + inbound_pipeline=inbound_pipeline, + outbound_pipeline=outbound_pipeline, + idempotency_repository=fake_idempotency_repo, + ) + + # Act + result = await service.receiveInbound(_make_inbound_cmd()) + await _drain_outbound_tasks(service) + + # Assert 幂等记录被回退为 failed（与 cancelled/interrupted 相反） + statuses = [call.args[1] for call in fake_idempotency_repo.updateIdempotencyStatus.call_args_list] + assert statuses[-1] == "failed" + # ERROR 日志透传真实失败原因 + error_logs = [ + call for call in fake_logger.error.call_args_list + if "outbound pipeline failed" in str(call) + ] + assert error_logs + error_call_args = error_logs[0].args + # 错误消息含 worker_error / uid contains invalid characters + error_msg = error_call_args[0] if error_call_args else "" + assert "worker_error" in error_msg + assert "uid contains invalid characters" in error_msg + assert result.ack_decision == "pending" + + @pytest.mark.asyncio + async def test_empty_output_agent_run_still_triggers_outbound_failure_handling( + self, + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + fake_idempotency_repo, + ): + # Arrange AgentRun completed 但输出为空（非 SKIP 类错误）→ 出站 + # 返回 ok=False 携带 AgentRunEmptyOutputError，应仍按 delivery + # failure 处理（与 AgentRunFailedError 语义一致） + inbound_pipeline = AsyncMock() + + async def _set_agent_run_with_idempotency(ctx): + ctx.agent_run_id = "run-empty-1" + ctx.idempotency_record_id = 42 + return (True, None) + + inbound_pipeline.run.side_effect = _set_agent_run_with_idempotency + outbound_pipeline = AsyncMock() + err = AgentRunEmptyOutputError( + run_id="run-empty-1", + agent_id="default-chatbot", + trace_id="trace-1", + ) + outbound_pipeline.run.return_value = (False, err) + streaming_config = MagicMock() + streaming_config.streaming_enabled = False + fake_config.getStreamingConfig.return_value = streaming_config + service = _make_service( + fake_logger, + fake_config, + fake_tracer, + fake_transaction_port, + inbound_pipeline=inbound_pipeline, + outbound_pipeline=outbound_pipeline, + idempotency_repository=fake_idempotency_repo, + ) + + # Act + result = await service.receiveInbound(_make_inbound_cmd()) + await _drain_outbound_tasks(service) + + # Assert 幂等记录被回退为 failed + statuses = [call.args[1] for call in fake_idempotency_repo.updateIdempotencyStatus.call_args_list] + assert statuses[-1] == "failed" + # 不应有 SKIP 日志 + skip_logs = [ + call for call in fake_logger.info.call_args_list + if "outbound pipeline skipped" in str(call) + ] + assert not skip_logs + assert result.ack_decision == "pending" + + @pytest.mark.unit class TestReceiveWebhook: """receiveWebhook 行为测试。"""