AiDD MCP Server

from .state import state from datetime import datetime from pathlib import Path from typing import Any, Dict, List import json class TodoStore: """Manages todo persistence and operations.""" def __init__(self): self._cached_store = None self._last_workspace = None @property def workspace_path(self) -> Path: """Get the current workspace directory.""" return Path(state.allowed_directory) @property def todos_file_path(self) -> Path: """Get the path to the global todos file.""" return state.config_dir / "todos.json" def _detect_workspace_change(self) -> bool: """Check if workspace has changed since last access.""" current_workspace = str(self.workspace_path) if self._last_workspace != current_workspace: self._last_workspace = current_workspace self._cached_store = None return True return False def _load_store(self) -> Dict[str, Any]: """Load todos from file with caching.""" self._detect_workspace_change() if self._cached_store is not None: return self._cached_store workspace_key = str(self.workspace_path) if not self.todos_file_path.exists(): self._cached_store = {"lastModified": datetime.now().isoformat(), "todos": []} return self._cached_store try: with open(self.todos_file_path, "r", encoding="utf-8") as f: global_data = json.load(f) workspace_data = global_data.get(workspace_key, {}) self._cached_store = {"lastModified": workspace_data.get("lastModified", datetime.now().isoformat()), "todos": workspace_data.get("todos", [])} return self._cached_store except (json.JSONDecodeError, IOError, OSError): # Return empty store if file is corrupted self._cached_store = {"lastModified": datetime.now().isoformat(), "todos": []} return self._cached_store def _save_store(self, store: Dict[str, Any]) -> None: """Save todos to file atomically.""" # Ensure the ~/.skydeckai-code directory exists self.todos_file_path.parent.mkdir(exist_ok=True) workspace_key = str(self.workspace_path) # Load existing global data global_data = {} if self.todos_file_path.exists(): try: with open(self.todos_file_path, "r", encoding="utf-8") as f: global_data = json.load(f) except (json.JSONDecodeError, IOError, OSError): global_data = {} # Update the workspace data global_data[workspace_key] = store # Write to temporary file first (atomic write) temp_path = self.todos_file_path.with_suffix(".json.tmp") try: with open(temp_path, "w", encoding="utf-8") as f: json.dump(global_data, f, indent=2, ensure_ascii=False) # Atomic rename temp_path.replace(self.todos_file_path) # Update cache self._cached_store = store except Exception: # Clean up temp file if something went wrong if temp_path.exists(): temp_path.unlink() raise def _add_to_gitignore(self) -> None: """No longer needed since todos are stored in ~/.skydeckai-code/todo.json""" pass def read_todos(self) -> List[Dict[str, Any]]: """Read all todos from storage.""" store = self._load_store() return store["todos"] def write_todos(self, todos: List[Dict[str, Any]]) -> int: """Write todos to storage with validation.""" # Validate todos self._validate_todos(todos) # Process todos (add timestamps, etc.) processed_todos = [] current_time = datetime.now().isoformat() for todo in todos: processed_todo = dict(todo) # Ensure required fields have defaults processed_todo.setdefault("id", self._generate_id()) processed_todo.setdefault("status", "pending") processed_todo.setdefault("created_at", current_time) processed_todo["updated_at"] = current_time processed_todos.append(processed_todo) # Create new store new_store = {"lastModified": current_time, "todos": processed_todos} # Save to file self._save_store(new_store) return len(processed_todos) def update_todo(self, todo_id: str, updates: Dict[str, Any]) -> Dict[str, Any]: """Update a specific todo by ID.""" store = self._load_store() todos = store["todos"] # Find the todo to update todo_index = None original_todo = None for i, todo in enumerate(todos): if todo["id"] == todo_id: todo_index = i original_todo = todo break if todo_index is None or original_todo is None: raise ValueError(f"Todo with ID '{todo_id}' not found") # Check if status is changing to completed original_status = original_todo["status"] new_status = updates.get("status", original_status) is_completing = original_status != "completed" and new_status == "completed" # Create updated todo updated_todo = dict(todos[todo_index]) updated_todo.update(updates) updated_todo["updated_at"] = datetime.now().isoformat() # Replace the todo in the list updated_todos = todos.copy() updated_todos[todo_index] = updated_todo # Validate the entire list with the update self._validate_todos(updated_todos) # Save updated list new_store = {"lastModified": datetime.now().isoformat(), "todos": updated_todos} self._save_store(new_store) # Return status counts pending_count = sum(1 for t in updated_todos if t["status"] == "pending") in_progress_count = sum(1 for t in updated_todos if t["status"] == "in_progress") completed_count = sum(1 for t in updated_todos if t["status"] == "completed") result = {"updated_todo": updated_todo, "counts": {"pending": pending_count, "in_progress": in_progress_count, "completed": completed_count, "total": len(updated_todos)}} # If a todo was just completed, find and include the next pending todo if is_completing: next_todo = self._find_next_pending_todo(updated_todos, todo_index) if next_todo: result["next_todo"] = next_todo else: result["next_todo"] = None result["message"] = "All todos completed! No more pending tasks." return result def _find_next_pending_todo(self, todos: List[Dict[str, Any]], completed_index: int) -> Dict[str, Any] | None: """Find the next pending todo after the completed one in sequential order.""" # Look for the next pending todo starting from the position after the completed one for i in range(completed_index + 1, len(todos)): if todos[i]["status"] == "pending": return todos[i] # If no pending todo found after the completed one, look from the beginning # This handles cases where todos might be reordered or the completed one wasn't the first in-progress for i in range(completed_index): if todos[i]["status"] == "pending": return todos[i] # No pending todos found return None def _validate_todos(self, todos: List[Dict[str, Any]]) -> None: """Validate todos according to business rules.""" if not isinstance(todos, list): raise ValueError("Todos must be a list") # Check for required fields and collect IDs required_fields = {"id", "content", "status"} seen_ids = set() in_progress_count = 0 for i, todo in enumerate(todos): if not isinstance(todo, dict): raise ValueError(f"Todo at index {i} must be a dictionary") # Check required fields missing_fields = required_fields - set(todo.keys()) if missing_fields: raise ValueError(f"Todo at index {i} missing required fields: {missing_fields}") # Validate ID uniqueness todo_id = todo["id"] if not isinstance(todo_id, str) or not todo_id.strip(): raise ValueError(f"Todo at index {i} must have a non-empty string ID") if todo_id in seen_ids: raise ValueError(f"Duplicate todo ID found: {todo_id}") seen_ids.add(todo_id) # Validate status if todo["status"] not in ["pending", "in_progress", "completed"]: raise ValueError(f"Todo at index {i} has invalid status: {todo['status']}") if todo["status"] == "in_progress": in_progress_count += 1 # Validate content if not isinstance(todo["content"], str) or not todo["content"].strip(): raise ValueError(f"Todo at index {i} must have non-empty content") # Business rule: only one task can be in progress if in_progress_count > 1: raise ValueError("Only one task can be in_progress at a time") def _generate_id(self) -> str: """Generate a unique ID for a todo.""" import uuid return str(uuid.uuid4())[:8] # Global todo store instance todo_store = TodoStore()