(self, domain: str, task_id: str)
| 549 | task.setdefault("progress", {}).update(dict(progress_fields or {})) |
| 550 | |
| 551 | def get_domain_task(self, domain: str, task_id: str) -> Optional[Dict[str, Any]]: |
| 552 | domain_key = str(domain or "").strip().lower() |
| 553 | task_key = str(task_id or "").strip() |
| 554 | if not domain_key or not task_key: |
| 555 | return None |
| 556 | with _domain_lock: |
| 557 | task = _domain_tasks.get(domain_key, {}).get(task_key) |
| 558 | return self._domain_task_snapshot(task) if task else None |
| 559 | |
| 560 | def list_domain_tasks(self, domain: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: |
| 561 | safe_limit = max(1, min(500, int(limit or 100))) |
no test coverage detected