1  # Copyright (c) 2024-2026 Tencent Zhuque Lab. All rights reserved.
  2  #
  3  # Licensed under the Apache License, Version 2.0 (the "License");
  4  # you may not use this file except in compliance with the License.
  5  # You may obtain a copy of the License at
  6  #
  7  #     http://www.apache.org/licenses/LICENSE-2.0
  8  #
  9  # Unless required by applicable law or agreed to in writing, software
 10  # distributed under the License is distributed on an "AS IS" BASIS,
 11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 12  # See the License for the specific language governing permissions and
 13  # limitations under the License.
 14  #
 15  # Requirement: Any integration or derivative work must explicitly attribute
 16  # Tencent Zhuque Lab (https://github.com/Tencent/AI-Infra-Guard) in its
 17  # documentation or user interface, as detailed in the NOTICE file.
 18  
 19  """
 20  Todo management module - Uses <repo>/.agent-scan/todos.json to cache Agent Todo
 21  支持加载、校验与持久化
 22  """
 23  import os
 24  import json
 25  from typing import List, Optional, Dict, Any
 26  from dataclasses import dataclass, asdict
 27  from enum import Enum
 28  
 29  
 30  class TodoStatus(str, Enum):
 31      """Todo 状态枚举"""
 32      PENDING = "pending"
 33      IN_PROGRESS = "in_progress"
 34      COMPLETED = "completed"
 35      CANCELLED = "cancelled"
 36  
 37  
 38  @dataclass
 39  class TodoItem:
 40      """Todo 条目"""
 41      id: str
 42      content: str
 43      status: TodoStatus
 44      priority: int = 0
 45      metadata: Optional[Dict[str, Any]] = None
 46      
 47      def to_dict(self) -> Dict[str, Any]:
 48          """转换为字典"""
 49          data = {
 50              "id": self.id,
 51              "content": self.content,
 52              "status": self.status.value if isinstance(self.status, TodoStatus) else self.status,
 53              "priority": self.priority,
 54          }
 55          if self.metadata:
 56              data["metadata"] = self.metadata
 57          return data
 58      
 59      @classmethod
 60      def from_dict(cls, data: Dict[str, Any]) -> "TodoItem":
 61          """从字典创建"""
 62          status = data.get("status", TodoStatus.PENDING)
 63          if isinstance(status, str):
 64              status = TodoStatus(status)
 65          
 66          return cls(
 67              id=data["id"],
 68              content=data["content"],
 69              status=status,
 70              priority=data.get("priority", 0),
 71              metadata=data.get("metadata"),
 72          )
 73  
 74  
 75  class TodoManager:
 76      """Todo 管理器"""
 77      
 78      def __init__(self, root_folder: str):
 79          """
 80          初始化 Todo 管理器
 81          
 82          Args:
 83              root_folder: 项目根目录
 84          """
 85          self.root_folder = root_folder
 86          self.cache_dir = os.path.join(root_folder, ".agent-scan")
 87          self.todo_file = os.path.join(self.cache_dir, "todos.json")
 88          self._todos: List[TodoItem] = []
 89          self._loaded = False
 90      
 91      def _ensure_cache_dir(self) -> None:
 92          """Ensure .agent-scan directory exists."""
 93          if not os.path.exists(self.cache_dir):
 94              os.makedirs(self.cache_dir, exist_ok=True)
 95      
 96      def load(self) -> List[TodoItem]:
 97          """
 98          加载 Todo 列表
 99          
100          Returns:
101              Todo 条目列表
102          """
103          if self._loaded:
104              return self._todos
105          
106          self._todos = []
107          
108          if os.path.exists(self.todo_file):
109              try:
110                  with open(self.todo_file, 'r', encoding='utf-8') as f:
111                      data = json.load(f)
112                  
113                  if isinstance(data, list):
114                      for item in data:
115                          try:
116                              self._todos.append(TodoItem.from_dict(item))
117                          except (KeyError, ValueError):
118                              continue
119                              
120              except (json.JSONDecodeError, IOError):
121                  self._todos = []
122          
123          self._loaded = True
124          return self._todos
125      
126      def save(self) -> None:
127          """持久化 Todo 列表"""
128          self._ensure_cache_dir()
129          
130          data = [todo.to_dict() for todo in self._todos]
131          
132          with open(self.todo_file, 'w', encoding='utf-8') as f:
133              json.dump(data, f, indent=2, ensure_ascii=False)
134      
135      def get_all(self) -> List[TodoItem]:
136          """获取所有 Todo"""
137          return self.load()
138      
139      def get_by_id(self, todo_id: str) -> Optional[TodoItem]:
140          """根据 ID 获取 Todo"""
141          self.load()
142          for todo in self._todos:
143              if todo.id == todo_id:
144                  return todo
145          return None
146      
147      def get_by_status(self, status: TodoStatus) -> List[TodoItem]:
148          """根据状态获取 Todo"""
149          self.load()
150          return [todo for todo in self._todos if todo.status == status]
151      
152      def add(self, todo: TodoItem) -> TodoItem:
153          """
154          添加 Todo
155          
156          Args:
157              todo: Todo 条目
158              
159          Returns:
160              添加的 Todo
161          """
162          self.load()
163          
164          # 检查 ID 是否已存在
165          existing = self.get_by_id(todo.id)
166          if existing:
167              raise ValueError(f"Todo with id '{todo.id}' already exists")
168          
169          self._todos.append(todo)
170          self.save()
171          return todo
172      
173      def update(self, todo_id: str, **updates) -> Optional[TodoItem]:
174          """
175          更新 Todo
176          
177          Args:
178              todo_id: Todo ID
179              **updates: 更新字段
180              
181          Returns:
182              更新后的 Todo，如果不存在返回 None
183          """
184          self.load()
185          
186          for i, todo in enumerate(self._todos):
187              if todo.id == todo_id:
188                  if "content" in updates:
189                      todo.content = updates["content"]
190                  if "status" in updates:
191                      status = updates["status"]
192                      if isinstance(status, str):
193                          status = TodoStatus(status)
194                      todo.status = status
195                  if "priority" in updates:
196                      todo.priority = updates["priority"]
197                  if "metadata" in updates:
198                      todo.metadata = updates["metadata"]
199                  
200                  self._todos[i] = todo
201                  self.save()
202                  return todo
203          
204          return None
205      
206      def remove(self, todo_id: str) -> bool:
207          """
208          删除 Todo
209          
210          Args:
211              todo_id: Todo ID
212              
213          Returns:
214              是否删除成功
215          """
216          self.load()
217          
218          for i, todo in enumerate(self._todos):
219              if todo.id == todo_id:
220                  self._todos.pop(i)
221                  self.save()
222                  return True
223          
224          return False
225      
226      def update_todos(self, todos: List[Dict[str, Any]]) -> List[TodoItem]:
227          """
228          批量更新/替换 Todo 列表
229          
230          Args:
231              todos: 新的 Todo 列表
232              
233          Returns:
234              更新后的 Todo 列表
235          """
236          self._todos = []
237          for item in todos:
238              try:
239                  self._todos.append(TodoItem.from_dict(item))
240              except (KeyError, ValueError):
241                  continue
242          
243          self.save()
244          self._loaded = True
245          return self._todos
246      
247      def clear(self) -> None:
248          """清空所有 Todo"""
249          self._todos = []
250          self.save()
251      
252      def validate_status_transition(self, current: TodoStatus, new: TodoStatus) -> bool:
253          """
254          验证状态转换是否合法
255          
256          Args:
257              current: 当前状态
258              new: 新状态
259              
260          Returns:
261              是否合法
262          """
263          # 定义合法的状态转换
264          valid_transitions = {
265              TodoStatus.PENDING: {TodoStatus.IN_PROGRESS, TodoStatus.CANCELLED},
266              TodoStatus.IN_PROGRESS: {TodoStatus.COMPLETED, TodoStatus.PENDING, TodoStatus.CANCELLED},
267              TodoStatus.COMPLETED: {TodoStatus.PENDING},  # 允许重新打开
268              TodoStatus.CANCELLED: {TodoStatus.PENDING},  # 允许恢复
269          }
270          
271          return new in valid_transitions.get(current, set())
272      
273      def get_summary(self) -> Dict[str, int]:
274          """
275          获取 Todo 统计摘要
276          
277          Returns:
278              各状态的数量统计
279          """
280          self.load()
281          
282          summary = {
283              "total": len(self._todos),
284              "pending": 0,
285              "in_progress": 0,
286              "completed": 0,
287              "cancelled": 0,
288          }
289          
290          for todo in self._todos:
291              status_key = todo.status.value if isinstance(todo.status, TodoStatus) else todo.status
292              if status_key in summary:
293                  summary[status_key] += 1
294          
295          return summary
296