(self, domain: str, task_id: str)
| 615 | return bool(task and task.get("pause_requested")) |
| 616 | |
| 617 | def request_domain_task_retry(self, domain: str, task_id: str) -> Optional[Dict[str, Any]]: |
| 618 | with _domain_lock: |
| 619 | task = _domain_tasks.get(str(domain or "").strip().lower(), {}).get(str(task_id or "").strip()) |
| 620 | if not task: |
| 621 | return None |
| 622 | task["retry_requested"] = True |
| 623 | return self._domain_task_snapshot(task) |
| 624 | |
| 625 | def try_acquire_domain_slot(self, domain: str, task_id: str) -> Tuple[bool, int, int]: |
| 626 | domain_key = str(domain or "").strip().lower() |
no test coverage detected