Handle task query (get task by ID or list all).
(self, data: Dict)
| 253 | } |
| 254 | |
| 255 | async def _handle_task(self, data: Dict) -> Dict: |
| 256 | """Handle task query (get task by ID or list all).""" |
| 257 | task_id = data.get("id") |
| 258 | |
| 259 | if task_id is not None: |
| 260 | # Get specific task |
| 261 | task = self.state.get_task(task_id) |
| 262 | if not task: |
| 263 | return {"status": "error", "message": f"Task {task_id} not found"} |
| 264 | return {"status": "ok", "task": task} |
| 265 | else: |
| 266 | # List all tasks |
| 267 | limit = data.get("limit", 20) |
| 268 | tasks = self.state.get_all_tasks()[:limit] |
| 269 | return {"status": "ok", "tasks": tasks} |
| 270 | |
| 271 | async def _handle_cancel(self, data: Dict) -> Dict: |
| 272 | """Handle task cancellation.""" |
nothing calls this directly
no test coverage detected