agent-Specialization/sub_agent/modules/todo_manager.py

# modules/todo_manager.py - TODO 列表管理

from __future__ import annotations

from copy import deepcopy
from typing import Dict, List, Any, Optional

try:
    from config import (
        TODO_MAX_TASKS,
        TODO_MAX_OVERVIEW_LENGTH,
        TODO_MAX_TASK_LENGTH,
    )
except ImportError:  # pragma: no cover
    import sys
    from pathlib import Path

    project_root = Path(__file__).resolve().parents[1]
    if str(project_root) not in sys.path:
        sys.path.insert(0, str(project_root))
    from config import (  # type: ignore
        TODO_MAX_TASKS,
        TODO_MAX_OVERVIEW_LENGTH,
        TODO_MAX_TASK_LENGTH,
    )


class TodoManager:
    """负责创建、更新和结束 TODO 列表"""

    MAX_TASKS = TODO_MAX_TASKS
    MAX_OVERVIEW_LENGTH = TODO_MAX_OVERVIEW_LENGTH
    MAX_TASK_LENGTH = TODO_MAX_TASK_LENGTH

    def __init__(self, context_manager):
        self.context_manager = context_manager

    def _get_current(self) -> Optional[Dict[str, Any]]:
        todo = getattr(self.context_manager, "todo_list", None)
        return deepcopy(todo) if todo else None

    def _save(self, todo: Optional[Dict[str, Any]]):
        self.context_manager.set_todo_list(todo)

    def _normalize_tasks(self, tasks: List[Any]) -> List[str]:
        normalized = []
        for item in tasks:
            title = ""
            if isinstance(item, dict):
                title = item.get("title", "")
            else:
                title = str(item)
            title = title.strip()
            if not title:
                continue
            normalized.append(title)
            if len(normalized) >= self.MAX_TASKS:
                break
        return normalized

    def create_todo_list(self, overview: str, tasks: List[Any]) -> Dict[str, Any]:
        current = self._get_current()
        if current and current.get("status") == "active":
            return {
                "success": False,
                "error": "已有进行中的 TODO 列表，请先完成或结束后再创建新的列表。"
            }

        overview = (overview or "").strip()
        if not overview:
            return {"success": False, "error": "任务概述不能为空。"}
        if len(overview) > self.MAX_OVERVIEW_LENGTH:
            return {
                "success": False,
                "error": f"任务概述过长（当前 {len(overview)} 字），请精简至 {self.MAX_OVERVIEW_LENGTH} 字以内。"
            }

        normalized_tasks = self._normalize_tasks(tasks or [])
        if not normalized_tasks:
            return {"success": False, "error": "需要至少提供一个任务。"}
        if len(tasks or []) > self.MAX_TASKS:
            return {
                "success": False,
                "error": f"任务数量过多，最多允许 {self.MAX_TASKS} 个任务。"
            }

        for title in normalized_tasks:
            if len(title) > self.MAX_TASK_LENGTH:
                return {
                    "success": False,
                    "error": f"任务「{title}」过长，请控制在 {self.MAX_TASK_LENGTH} 字以内。"
                }

        todo = {
            "overview": overview,
            "tasks": [
                {
                    "index": idx,
                    "title": title,
                    "status": "pending"
                }
                for idx, title in enumerate(normalized_tasks, start=1)
            ],
            "status": "active",
            "forced_finish": False,
            "forced_reason": None
        }

        self._save(todo)
        return {
            "success": True,
            "message": "待办列表已创建。请先完成某项任务，再调用待办工具将其标记完成。",
            "todo_list": todo
        }

    def update_task_status(self, task_index: int, completed: bool) -> Dict[str, Any]:
        todo = self._get_current()
        if not todo:
            return {"success": False, "error": "当前没有待办列表，请先创建。"}
        if todo.get("status") in {"completed", "closed"}:
            return {"success": False, "error": "待办列表已结束，无法继续修改。"}

        if not isinstance(task_index, int):
            return {"success": False, "error": "task_index 必须是数字。"}
        if task_index < 1 or task_index > len(todo["tasks"]):
            return {"success": False, "error": f"task_index 超出范围（1-{len(todo['tasks'])}）。"}

        task = todo["tasks"][task_index - 1]
        new_status = "done" if completed else "pending"
        if task["status"] == new_status:
            return {
                "success": True,
                "message": "任务状态未发生变化。",
                "todo_list": todo
            }

        task["status"] = new_status

        self._save(todo)
        return {
            "success": True,
            "message": f"任务 task{task_index} 已标记为 {'完成' if completed else '未完成'}。",
            "todo_list": todo
        }

    def finish_todo(self, reason: Optional[str] = None) -> Dict[str, Any]:
        todo = self._get_current()
        if not todo:
            return {"success": False, "error": "当前没有待办列表。"}

        if todo.get("status") in {"completed", "closed"}:
            return {
                "success": True,
                "message": "待办列表已结束，无需重复操作。",
                "todo_list": todo
            }

        all_done = all(task["status"] == "done" for task in todo["tasks"])
        if all_done:
            todo["status"] = "completed"
            todo["forced_finish"] = False
            todo["forced_reason"] = None
            self._save(todo)
            system_note = "✅ TODO 列表中的所有任务已完成，可以整理成果并向用户汇报。"
            return {
                "success": True,
                "message": "所有任务已完成，待办列表已结束。",
                "todo_list": todo,
                "system_note": system_note
            }

        remaining = [
            f"task{task['index']}"
            for task in todo["tasks"]
            if task["status"] != "done"
        ]
        return {
            "success": False,
            "requires_confirmation": True,
            "message": "仍有未完成的任务，确认要提前结束吗？",
            "remaining": remaining,
            "todo_list": todo
        }

    def confirm_finish(self, confirm: bool, reason: Optional[str] = None) -> Dict[str, Any]:
        todo = self._get_current()
        if not todo:
            return {"success": False, "error": "当前没有待办列表。"}
        if todo.get("status") in {"completed", "closed"}:
            return {
                "success": True,
                "message": "待办列表已结束，无需重复操作。",
                "todo_list": todo
            }

        if not confirm:
            return {
                "success": True,
                "message": "已取消结束待办列表，继续执行剩余任务。",
                "todo_list": todo
            }

        todo["status"] = "closed"
        todo["forced_finish"] = True
        todo["forced_reason"] = (reason or "").strip() or None
        self._save(todo)

        system_note = "⚠️ TODO 列表在任务未全部完成的情况下被结束，请在总结中说明原因。"
        self.context_manager.add_conversation("system", system_note)
        return {
            "success": True,
            "message": "待办列表已强制结束。",
            "todo_list": todo,
            "system_note": system_note
        }

    def get_snapshot(self) -> Optional[Dict[str, Any]]:
        return self._get_current()