agent-Specialization/modules/sub_agent/state.py

"""子智能体状态管理 Mixin。"""

from __future__ import annotations

import asyncio
import json
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

from utils.logger import setup_logger
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:
    """提供子智能体任务状态加载、保存、刷新与结果落盘能力。"""

    state_file: Path
    tasks: Dict[str, Dict[str, Any]]
    conversation_agents: Dict[str, List[int]]
    _running_tasks: Dict[str, Any]
    _state_lock: Any

    def _load_state(self):
        with self._state_lock:
            if not self.state_file.exists():
                self.tasks = {}
                self.conversation_agents = {}
                return
            try:
                data = json.loads(self.state_file.read_text(encoding="utf-8"))
                loaded_tasks = data.get("tasks", {})
                loaded_agents = data.get("conversation_agents", {})
            except json.JSONDecodeError:
                logger.warning("子智能体状态文件损坏，已忽略。")
                self.tasks = {}
                self.conversation_agents = {}
                return

            runtime_only_keys = {"_stdout_lines"}
            merged_tasks: Dict[str, Dict] = {}
            for task_id, task in loaded_tasks.items():
                existing = self.tasks.get(task_id)
                if existing:
                    for key in runtime_only_keys:
                        if key in existing:
                            task[key] = existing[key]
                merged_tasks[task_id] = task
            self.tasks = merged_tasks
            self.conversation_agents = loaded_agents

            # 恢复多智能体运行态（如果状态文件包含）
            try:
                from modules.multi_agent.state import MultiAgentState

                manager = self
                multi_agent_states = getattr(manager, "multi_agent_states", None)
                if multi_agent_states is not None and isinstance(multi_agent_states, dict):
                    loaded_ma_states = data.get("multi_agent_states", {})
                    for conv_id, snapshot in loaded_ma_states.items():
                        try:
                            if isinstance(snapshot, dict):
                                # 关键：不要覆盖内存中已存在的 MultiAgentState，
                                # 否则 SubAgentTask 持有的旧引用上的 pending_master_messages
                                # 会被新的空 state 覆盖，导致子智能体输出丢失。
                                if conv_id in multi_agent_states:
                                    ma_debug(
                                        "load_state_skip_existing_ma_state",
                                        conversation_id=conv_id,
                                        existing_state_id=id(multi_agent_states[conv_id]),
                                    )
                                    continue
                                multi_agent_states[conv_id] = MultiAgentState.from_snapshot(snapshot)
                                ma_debug(
                                    "load_state_restore_ma_state",
                                    conversation_id=conv_id,
                                    state_id=id(multi_agent_states[conv_id]),
                                )
                        except Exception as exc:
                            logger.warning(f"恢复多智能体状态失败 {conv_id}: {exc}")
            except Exception as exc:
                logger.warning(f"加载多智能体状态失败: {exc}")

            if self.tasks:
                migrated = False
                for task in self.tasks.values():
                    if task.get("parent_conversation_id"):
                        continue
                    candidate = task.get("conversation_id") or (task.get("service_payload") or {}).get("parent_conversation_id")
                    if candidate:
                        task["parent_conversation_id"] = candidate
                        migrated = True
                if migrated:
                    self._save_state_unsafe()

    def _save_state(self):
        with self._state_lock:
            self._save_state_unsafe()

    def _save_state_unsafe(self):
        payload = {
            "tasks": self.tasks,
            "conversation_agents": self.conversation_agents,
        }
        # 多智能体运行态持久化
        try:
            manager = self
            multi_agent_states = getattr(manager, "multi_agent_states", None)
            if multi_agent_states:
                payload["multi_agent_states"] = {
                    conv_id: state.to_snapshot()
                    for conv_id, state in multi_agent_states.items()
                    if isinstance(state, object) and hasattr(state, "to_snapshot")
                }
        except Exception as exc:
            logger.warning(f"保存多智能体状态失败: {exc}")
        try:
            self.state_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
        except Exception as exc:
            logger.warning(f"保存子智能体状态失败: {exc}")

    def _check_task_status(self, task: Dict) -> Dict:
        """检查任务状态，读取输出文件。"""
        task_id = task["task_id"]
        output_file = Path(task.get("output_file", ""))
        if not output_file.exists():
            running_task = self._running_tasks.get(task_id)
            if running_task and running_task.done():
                try:
                    running_task.result(timeout=0)
                except Exception:
                    pass
                task["status"] = "failed"
                task["updated_at"] = time.time()
            return {"status": task.get("status", "running"), "task_id": task_id}

        try:
            output = json.loads(output_file.read_text(encoding="utf-8"))
        except Exception as exc:
            logger.warning(f"[check_task_status] output 文件解析失败: {exc}")
            task["status"] = "failed"
            task["updated_at"] = time.time()
            return {"success": False, "status": "failed", "task_id": task_id, "message": f"输出文件解析失败: {exc}"}

        raw_success = output.get("success")
        success = bool(raw_success)
        summary = output.get("summary", "")
        stats = output.get("stats", {})
        elapsed_seconds = self._compute_elapsed_seconds(task)

        ma_debug(
            "check_task_status",
            task_id=task_id,
            agent_id=task.get("agent_id"),
            multi_agent_mode=task.get("multi_agent_mode"),
            output_status=output.get("status"),
            output_success=success,
        )

        # 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"
        elif output.get("max_turns_exceeded"):
            status = "failed"
            summary = f"任务执行超过最大轮次限制。{summary}"
        elif success:
            status = "completed"
        else:
            status = "failed"

        ma_debug("check_task_status_result", task_id=task_id, agent_id=task.get("agent_id"), status=status)
        task["status"] = status
        task["updated_at"] = time.time()
        if status == "completed" and elapsed_seconds is not None:
            task["elapsed_seconds"] = elapsed_seconds
            task["runtime_seconds"] = elapsed_seconds

        agent_id = task.get("agent_id")
        task_summary = task.get("summary")
        deliverables_dir = task.get("deliverables_dir")
        stats_summary = self._build_stats_summary(stats)

        if status == "completed":
            system_message = self._compose_sub_agent_message(
                prefix=f"✅ 子智能体{agent_id} 任务摘要：{task_summary} 已完成。",
                stats_summary=stats_summary,
                summary=summary,
                deliverables_dir=deliverables_dir,
                duration_seconds=elapsed_seconds,
            )
        elif status == "timeout":
            system_message = self._compose_sub_agent_message(
                prefix=f"⏱️ 子智能体{agent_id} 任务摘要：{task_summary} 超时未完成。",
                stats_summary=stats_summary,
                summary=summary,
            )
        else:
            system_message = self._compose_sub_agent_message(
                prefix=f"❌ 子智能体{agent_id} 任务摘要：{task_summary} 执行失败。",
                stats_summary=stats_summary,
                summary=summary,
            )

        result = {
            "success": success,
            "status": status,
            "task_id": task_id,
            "agent_id": agent_id,
            "message": summary,
            "deliverables_dir": deliverables_dir,
            "stats": stats,
            "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"]
        running_task = self._running_tasks.pop(task_id, None)
        if running_task and not running_task.done():
            running_task.cancel()
            deadline = time.time() + 5
            while not running_task.done() and time.time() < deadline:
                time.sleep(0.05)

        task["status"] = "timeout"
        task["updated_at"] = time.time()

        stats = {}
        stats_file = Path(task.get("stats_file", ""))
        if stats_file.exists():
            try:
                stats = json.loads(stats_file.read_text(encoding="utf-8"))
            except Exception:
                stats = {}
        stats_summary = self._build_stats_summary(stats)
        system_message = self._compose_sub_agent_message(
            prefix=f"⏱️ 子智能体{task.get('agent_id')} 任务摘要：{task.get('summary')} 超时未完成。",
            stats_summary=stats_summary,
            summary="等待超时，子智能体已被终止。",
        )

        result = {
            "success": False,
            "status": "timeout",
            "task_id": task_id,
            "agent_id": task.get("agent_id"),
            "message": "等待超时，子智能体已被终止。",
            "stats": stats,
            "stats_summary": stats_summary,
            "system_message": system_message,
        }
        task["final_result"] = result
        self._save_state()
        return result

    def _mark_task_terminated(
        self,
        task: Dict[str, Any],
        *,
        message: str,
        system_message: Optional[str] = None,
        notified: bool = False,
    ) -> Dict[str, Any]:
        task["status"] = "terminated"
        task["updated_at"] = time.time()
        if notified:
            task["notified"] = True
        result = {
            "success": False,
            "status": "terminated",
            "task_id": task.get("task_id"),
            "agent_id": task.get("agent_id"),
            "message": message,
            "system_message": system_message or message,
        }
        task["final_result"] = result
        return result

    def _mark_task_done(
        self,
        task_id: str,
        success: bool,
        summary: str,
        runtime_seconds: int,
    ) -> None:
        task = self.tasks.get(task_id)
        if not task:
            return
        # 如果 terminate_sub_agent 已把任务标为 terminated，下渗异步任务自然结束时
        # 不要被“已完成/fulfilled覆盖为failed”值覆盖，否则 task 被碰成 failed 但实际上
        # 用户手动终止已发生，状态语义错误且“successful=false”。
        existing_status = task.get("status")
        if existing_status in TERMINAL_STATUSES.union({"terminated"}):
            ma_debug(
                "mark_task_done_skip_already_terminal",
                task_id=task_id,
                existing_status=existing_status,
            )
            return
        status = "completed" if success else "failed"
        ma_debug(
            "mark_task_done",
            task_id=task_id,
            agent_id=task.get("agent_id"),
            success=success,
            new_status=status,
        )
        task["status"] = status
        task["updated_at"] = time.time()
        task["runtime_seconds"] = runtime_seconds
        self._check_task_status(task)
        self._save_state()

    def _refresh_task_runtime_state(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """刷新单个任务运行态。"""
        status = task.get("status")
        task_id = task.get("task_id")
        agent_id = task.get("agent_id")
        multi_agent_flag = task.get("multi_agent_mode")
        ma_debug(
            "refresh_task_runtime_state_start",
            task_id=task_id,
            agent_id=agent_id,
            before_status=status,
            multi_agent_mode=multi_agent_flag,
        )
        if status in TERMINAL_STATUSES.union({"terminated"}):
            return {"status": status, "task_id": task_id}

        running_task = self._running_tasks.get(task_id) if task_id else None
        if running_task:
            if running_task.done():
                try:
                    running_task.result(timeout=0)
                except Exception:
                    pass
                # 多智能体模式：任务自然进入 idle 时不写输出文件，不应标记为失败
                if task.get("multi_agent_mode") and not Path(task.get("output_file", "")).exists():
                    task["status"] = "idle"
                    task["updated_at"] = time.time()
                    ma_debug("refresh_task_runtime_state_idle_no_output", task_id=task_id)
                    return {"status": "idle", "task_id": task_id}
                result = self._check_task_status(task)
                ma_debug("refresh_task_runtime_state_done_result", task_id=task_id, result=result)
                return result
            ma_debug("refresh_task_runtime_state_still_running", task_id=task_id)
            return {"status": "running", "task_id": task_id}

        output_file = Path(task.get("output_file", ""))
        if output_file.exists():
            result = self._check_task_status(task)
            ma_debug("refresh_task_runtime_state_output_result", task_id=task_id, result=result)
            return result

        # 多智能体模式：没有输出文件表示 idle，不强制清理
        if task.get("multi_agent_mode"):
            task["status"] = "idle"
            task["updated_at"] = time.time()
            ma_debug("refresh_task_runtime_state_idle_no_output_file", task_id=task_id)
            return {"status": "idle", "task_id": task_id}

        if self._should_force_cleanup_stale_task(task):
            return self._mark_task_terminated(
                task,
                message="子智能体疑似僵尸任务，已超时自动清理运行状态。",
                system_message="⚠️ 子智能体长时间未结束，系统已自动清理运行状态。",
                notified=True,
            )

        return self._mark_task_terminated(
            task,
            message="检测到子智能体任务已退出，已自动清理运行状态。",
            system_message="⚠️ 子智能体任务异常退出，系统已自动清理运行状态。",
            notified=True,
        )

    def reconcile_task_states(self, conversation_id: Optional[str] = None) -> int:
        """修正运行态任务状态，返回修正条目数。"""
        changed = 0
        ma_debug("reconcile_task_states_start", conversation_id=conversation_id, task_count=len(self.tasks))
        for task in self.tasks.values():
            if not isinstance(task, dict):
                continue
            if conversation_id and task.get("conversation_id") != conversation_id:
                continue
            before_status = task.get("status")
            before_notified = task.get("notified")
            self._refresh_task_runtime_state(task)
            after_status = task.get("status")
            if after_status != before_status or task.get("notified") != before_notified:
                ma_debug(
                    "reconcile_task_states_changed",
                    task_id=task.get("task_id"),
                    agent_id=task.get("agent_id"),
                    before_status=before_status,
                    after_status=after_status,
                )
                changed += 1
        if changed:
            self._save_state()
        ma_debug("reconcile_task_states_end", conversation_id=conversation_id, changed=changed)
        return changed

    def _should_force_cleanup_stale_task(self, task: Dict[str, Any]) -> bool:
        try:
            created_at = float(task.get("created_at") or 0)
        except (TypeError, ValueError):
            created_at = 0
        if created_at <= 0:
            return False
        # timeout_seconds 为 None 表示永久子智能体，不强制清理
        raw_timeout = task.get("timeout_seconds")
        if raw_timeout is None:
            return False
        from config import SUB_AGENT_DEFAULT_TIMEOUT
        timeout_seconds = int(raw_timeout or SUB_AGENT_DEFAULT_TIMEOUT or 0)
        timeout_seconds = max(timeout_seconds, 1)
        grace_seconds = 120
        elapsed = time.time() - created_at
        if elapsed <= (timeout_seconds + grace_seconds):
            return False
        output_file = Path(task.get("output_file", ""))
        if output_file.exists():
            return False
        progress_file = Path(task.get("progress_file", ""))
        if progress_file.exists():
            try:
                stale_span = time.time() - progress_file.stat().st_mtime
                if stale_span <= grace_seconds:
                    return False
            except Exception:
                pass
        return True