From c414d268b3abf04a1f02a122cf93815082370dcb Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 20 Jul 2026 01:04:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(sub-agent):=20=E4=BF=AE=E5=A4=8D=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E4=B8=AD=E5=BF=AB=E7=85=A7=E8=A2=AB=E8=AF=AF=E5=88=A4?= =?UTF-8?q?=20failed=20=E5=8F=8A=E9=94=99=E8=AF=AF=E4=BF=A1=E6=81=AF?= =?UTF-8?q?=E8=90=BD=E7=9B=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - output.json 停留在 running/idle 快照（success 为 None）时一律先走存活判定， 任何模式下都不把仍在运行的子智能体误判为 failed - 本进程 asyncio 句柄已结束但快照未更新：判 failed 并保留真实异常信息 - 传统模式无句柄（进程重启）：快照 600s 内未更新才按僵尸任务判 failed - _finalize_task 支持 error 字段，失败/超时/异常退出均落盘可读错误信息 --- modules/sub_agent/state.py | 100 +++++++++++++++++++++++++++++++++---- modules/sub_agent/task.py | 10 ++-- 2 files changed, 97 insertions(+), 13 deletions(-) diff --git a/modules/sub_agent/state.py b/modules/sub_agent/state.py index 13aafbb9..f3527972 100644 --- a/modules/sub_agent/state.py +++ b/modules/sub_agent/state.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import json import time from pathlib import Path @@ -12,6 +13,10 @@ from modules.multi_agent.debug_logger import ma_debug logger = setup_logger(__name__) TERMINAL_STATUSES = {"completed", "failed", "timeout"} +# output.json 停留在 running/idle 快照（success 为 None）时的僵尸判定阈值： +# 传统模式且本进程无运行句柄（如进程重启后），快照超过该时长未更新才允许判 failed， +# 避免把仍在运行的子智能体误判为失败。 +STALE_RUNNING_SNAPSHOT_SECONDS = 600 class SubAgentStateMixin: @@ -144,7 +149,8 @@ class SubAgentStateMixin: task["updated_at"] = time.time() return {"success": False, "status": "failed", "task_id": task_id, "message": f"输出文件解析失败: {exc}"} - success = output.get("success", False) + raw_success = output.get("success") + success = bool(raw_success) summary = output.get("summary", "") stats = output.get("stats", {}) elapsed_seconds = self._compute_elapsed_seconds(task) @@ -158,14 +164,12 @@ class SubAgentStateMixin: output_success=success, ) - # 多智能体模式：output 中 status 为 running/idle 表示子智能体仍在运行或 - # 本轮结束但上下文保留、可继续接收消息，都不是失败/完成。 - # 此时不生成 final_result，只保持原状态，避免 reconcile 把运行中误判为失败。 - if task.get("multi_agent_mode") and output.get("status") in {"running", "idle"}: - task["status"] = output["status"] - task["updated_at"] = time.time() - ma_debug("check_task_status_keep_alive", task_id=task_id, status=output["status"]) - return {"status": output["status"], "task_id": task_id} + # output.json 在子智能体存活期间是运行中周期快照（success 为 None， + # status 为 running/idle），最终写入必然带 success 布尔值。 + # 遇到运行中快照一律先走存活判定：任何模式下都不能把仍在运行的 + # 子智能体误判为 failed（传统模式曾因此被 poll_updates 误杀）。 + if raw_success is None and output.get("status") in {"running", "idle"}: + return self._handle_running_snapshot(task, output_file, str(output.get("status"))) if output.get("timeout"): status = "timeout" @@ -221,12 +225,90 @@ class SubAgentStateMixin: "stats_summary": stats_summary, "system_message": system_message, } + if status in {"failed", "timeout"}: + error_text = str(output.get("error") or "").strip() + if error_text: + result["error"] = error_text if status == "completed" and elapsed_seconds is not None: result["elapsed_seconds"] = elapsed_seconds result["runtime_seconds"] = elapsed_seconds task["final_result"] = result return result + def _handle_running_snapshot(self, task: Dict, output_file: Path, output_status: str) -> Dict: + """处理 output.json 停留在 running/idle 快照（success 为 None）的状态判定。 + + - 本进程内 asyncio 句柄仍存活：任务确实在跑，保持 running（所有模式）； + - 句柄已结束但快照未更新：子智能体异常退出、未来得及写最终结果，判 failed 并记录异常； + - 多智能体模式无句柄：running/idle 均为正常存活态，保持原状； + - 传统模式无句柄（如进程重启）：快照新鲜视为仍在运行（可能由同数据目录的其他 + 进程写入），超过 STALE_RUNNING_SNAPSHOT_SECONDS 未更新按僵尸任务判 failed。 + """ + task_id = task["task_id"] + running_task = self._running_tasks.get(task_id) + if running_task is not None: + if not running_task.done(): + task["status"] = "running" + task["updated_at"] = time.time() + ma_debug("check_task_status_keep_alive", task_id=task_id, status="running") + return {"status": "running", "task_id": task_id} + # 句柄已结束但输出仍是运行中快照：子智能体异常退出 + error_text = "子智能体异常退出，未写入最终执行结果" + try: + running_task.result(timeout=0) + except asyncio.CancelledError: + error_text = "子智能体任务被取消，未写入最终执行结果" + except BaseException as exc: # 需要保留真实异常信息 + detail = str(exc) or exc.__class__.__name__ + error_text = f"子智能体异常退出：{detail}" + ma_debug("check_task_status_crashed_snapshot", task_id=task_id, error=error_text) + return self._fail_task_from_snapshot(task, error_text) + if task.get("multi_agent_mode"): + # 多智能体模式：running/idle 表示仍在运行或本轮结束但上下文保留 + task["status"] = output_status + task["updated_at"] = time.time() + ma_debug("check_task_status_keep_alive", task_id=task_id, status=output_status) + return {"status": output_status, "task_id": task_id} + # 传统模式无运行句柄：按快照新旧程度兜底 + try: + snapshot_age = time.time() - output_file.stat().st_mtime + except OSError: + snapshot_age = float("inf") + if snapshot_age < STALE_RUNNING_SNAPSHOT_SECONDS: + task["status"] = "running" + task["updated_at"] = time.time() + ma_debug( + "check_task_status_keep_alive_no_handle", + task_id=task_id, + snapshot_age=int(snapshot_age), + ) + return {"status": "running", "task_id": task_id} + error_text = ( + f"子智能体输出停留在运行中快照且已 {int(snapshot_age)} 秒未更新，" + "疑似进程中断或任务崩溃" + ) + ma_debug("check_task_status_stale_snapshot", task_id=task_id, snapshot_age=int(snapshot_age)) + return self._fail_task_from_snapshot(task, error_text) + + def _fail_task_from_snapshot(self, task: Dict, error_text: str) -> Dict: + """将无法恢复的运行中快照任务标记为 failed，并落盘可读的错误信息。""" + task["status"] = "failed" + task["updated_at"] = time.time() + result = { + "success": False, + "status": "failed", + "task_id": task.get("task_id"), + "agent_id": task.get("agent_id"), + "message": error_text, + "error": error_text, + "system_message": ( + f"❌ 子智能体{task.get('agent_id')} 任务摘要：{task.get('summary')} " + f"执行失败。{error_text}" + ), + } + task["final_result"] = result + return result + def _handle_timeout(self, task: Dict) -> Dict: """处理任务超时。""" task_id = task["task_id"] diff --git a/modules/sub_agent/task.py b/modules/sub_agent/task.py index 29e7bd0c..21af2fb1 100644 --- a/modules/sub_agent/task.py +++ b/modules/sub_agent/task.py @@ -910,10 +910,11 @@ class SubAgentTask: async def _write_finish(self, args: Dict[str, Any], elapsed: float) -> None: success = bool(args.get("success", False)) summary = str(args.get("summary") or "").strip() - self._finalize_task(success, summary, elapsed) + error = None if success else (summary or "子智能体报告执行失败，未说明原因") + self._finalize_task(success, summary, elapsed, error=error) async def _write_timeout(self, elapsed: float) -> None: - self._finalize_task(False, "任务超时未完成", elapsed, timeout=True) + self._finalize_task(False, "任务超时未完成", elapsed, timeout=True, error="任务超时未完成") async def _write_failure(self, message: str, *, max_turns_exceeded: bool = False, timeout: bool = False) -> None: elapsed = time.time() - (self.stats["runtime_start"] / 1000) @@ -928,7 +929,7 @@ class SubAgentTask: idle=self._idle, cancelled=self._cancelled, ) - self._finalize_task(False, message, elapsed, max_turns_exceeded=max_turns_exceeded, timeout=timeout) + self._finalize_task(False, message, elapsed, max_turns_exceeded=max_turns_exceeded, timeout=timeout, error=message) def _persist_conversation(self, *, partial_summary: str = "") -> None: """每轮结束后立即落盘子智能体对话，避免跑完了才存一次导致中间状态丢失。""" @@ -1041,7 +1042,7 @@ class SubAgentTask: if self.multi_agent_state: self.multi_agent_state.mark_status(self.agent_id, "idle") - def _finalize_task(self, success: bool, summary: str, elapsed: float, *, max_turns_exceeded: bool = False, timeout: bool = False) -> None: + def _finalize_task(self, success: bool, summary: str, elapsed: float, *, max_turns_exceeded: bool = False, timeout: bool = False, error: Optional[str] = None) -> None: runtime_seconds = int(elapsed) ma_debug( "sub_agent_finalize_task", @@ -1065,6 +1066,7 @@ class SubAgentTask: "success": success, "status": status, "summary": summary, + "error": error, "timeout": timeout, "max_turns_exceeded": max_turns_exceeded, "stats": {**self.stats, "runtime_seconds": runtime_seconds, "turn_count": self.stats.get("turn_count", 0)},