| 120 | |
| 121 | # -- TaskManager: persistent task board with optional worktree binding -- |
| 122 | class TaskManager: |
| 123 | def __init__(self, tasks_dir: Path): |
| 124 | self.dir = tasks_dir |
| 125 | self.dir.mkdir(parents=True, exist_ok=True) |
| 126 | self._next_id = self._max_id() + 1 |
| 127 | |
| 128 | def _max_id(self) -> int: |
| 129 | ids = [] |
| 130 | for f in self.dir.glob("task_*.json"): |
| 131 | try: |
| 132 | ids.append(int(f.stem.split("_")[1])) |
| 133 | except Exception: |
| 134 | pass |
| 135 | return max(ids) if ids else 0 |
| 136 | |
| 137 | def _path(self, task_id: int) -> Path: |
| 138 | return self.dir / f"task_{task_id}.json" |
| 139 | |
| 140 | def _load(self, task_id: int) -> dict: |
| 141 | path = self._path(task_id) |
| 142 | if not path.exists(): |
| 143 | raise ValueError(f"Task {task_id} not found") |
| 144 | return json.loads(path.read_text()) |
| 145 | |
| 146 | def _save(self, task: dict): |
| 147 | self._path(task["id"]).write_text(json.dumps(task, indent=2)) |
| 148 | |
| 149 | def create(self, subject: str, description: str = "") -> str: |
| 150 | task = { |
| 151 | "id": self._next_id, |
| 152 | "subject": subject, |
| 153 | "description": description, |
| 154 | "status": "pending", |
| 155 | "owner": "", |
| 156 | "worktree": "", |
| 157 | "blockedBy": [], |
| 158 | "created_at": time.time(), |
| 159 | "updated_at": time.time(), |
| 160 | } |
| 161 | self._save(task) |
| 162 | self._next_id += 1 |
| 163 | return json.dumps(task, indent=2) |
| 164 | |
| 165 | def get(self, task_id: int) -> str: |
| 166 | return json.dumps(self._load(task_id), indent=2) |
| 167 | |
| 168 | def exists(self, task_id: int) -> bool: |
| 169 | return self._path(task_id).exists() |
| 170 | |
| 171 | def update(self, task_id: int, status: str = None, owner: str = None) -> str: |
| 172 | task = self._load(task_id) |
| 173 | if status: |
| 174 | if status not in ("pending", "in_progress", "completed"): |
| 175 | raise ValueError(f"Invalid status: {status}") |
| 176 | task["status"] = status |
| 177 | if owner is not None: |
| 178 | task["owner"] = owner |
| 179 | task["updated_at"] = time.time() |
no outgoing calls
no test coverage detected