(task_id: str, owner: str = "agent")
| 108 | |
| 109 | |
| 110 | def claim_task(task_id: str, owner: str = "agent") -> str: |
| 111 | task = load_task(task_id) |
| 112 | if task.status != "pending": |
| 113 | return f"Task {task_id} is {task.status}, cannot claim" |
| 114 | if not can_start(task_id): |
| 115 | deps = [d for d in task.blockedBy |
| 116 | if not _task_path(d).exists() or load_task(d).status != "completed"] |
| 117 | return f"Blocked by: {deps}" |
| 118 | task.owner = owner |
| 119 | task.status = "in_progress" |
| 120 | save_task(task) |
| 121 | print(f" \033[36m[claim] {task.subject} → in_progress (owner: {owner})\033[0m") |
| 122 | return f"Claimed {task.id} ({task.subject})" |
| 123 | |
| 124 | |
| 125 | def complete_task(task_id: str) -> str: |
no test coverage detected