(task_id: str, owner: str = "agent")
| 104 | |
| 105 | |
| 106 | def claim_task(task_id: str, owner: str = "agent") -> str: |
| 107 | task = load_task(task_id) |
| 108 | if task.status != "pending": |
| 109 | return f"Task {task_id} is {task.status}, cannot claim" |
| 110 | if task.owner: |
| 111 | return f"Task {task_id} already owned by {task.owner}" |
| 112 | if not can_start(task_id): |
| 113 | deps = [d for d in task.blockedBy |
| 114 | if _task_path(d).exists() and load_task(d).status != "completed"] |
| 115 | missing = [d for d in task.blockedBy if not _task_path(d).exists()] |
| 116 | parts = [] |
| 117 | if deps: parts.append(f"blocked by: {deps}") |
| 118 | if missing: parts.append(f"missing deps: {missing}") |
| 119 | return "Cannot start — " + ", ".join(parts) |
| 120 | task.owner = owner |
| 121 | task.status = "in_progress" |
| 122 | save_task(task) |
| 123 | print(f" \033[36m[claim] {task.subject} → in_progress\033[0m") |
| 124 | return f"Claimed {task.id} ({task.subject})" |
| 125 | |
| 126 | |
| 127 | def complete_task(task_id: str) -> str: |
no test coverage detected