(self, task_id: str, error: str, error_type: Optional[str] = None)
| 125 | task.save_result_path = save_result_path |
| 126 | task.result_png = result_png |
| 127 | task.usage = usage |
| 128 | task.result_data = result_data |
| 129 | |
| 130 | if result_png is not None: |
| 131 | self._evict_old_result_png_unlocked() |
| 132 | |
| 133 | self.completed_tasks += 1 |
| 134 | self._emit_queue_metrics_unlocked() |
| 135 | |
| 136 | def fail_task(self, task_id: str, error: str, error_type: Optional[str] = None): |
| 137 | with self._lock: |
| 138 | if task_id not in self._tasks: |
| 139 | logger.warning(f"Task {task_id} not found for failure") |
| 140 | return |
| 141 | |
| 142 | task = self._tasks[task_id] |
| 143 | task.status = TaskStatus.FAILED |
no test coverage detected