Method register_domain_task

src/web/task_manager.py:498–517

(
        self,
        *,
        domain: str,
        task_id: str,
        task_type: str,
        payload: Optional[Dict[str, Any]] = None,
        progress: Optional[Dict[str, Any]] = None,
        max_retries: int = 0,
    )

            return len(_domain_running.get(domain_key, set()))

    def register_domain_task(
        self,
        *,
        domain: str,
        task_id: str,
        task_type: str,
        payload: Optional[Dict[str, Any]] = None,
        progress: Optional[Dict[str, Any]] = None,
        max_retries: int = 0,
    ) -> Dict[str, Any]:
        with _domain_lock:
            task = self._ensure_domain_task_locked(
                domain=domain,
                task_id=task_id,
                task_type=task_type,
                payload=payload,
                progress=progress,
            )
            task["max_retries"] = max(0, int(max_retries or 0))
            return self._domain_task_snapshot(task)

    def update_domain_task(self, domain: str, task_id: str, **fields) -> Optional[Dict[str, Any]]:
        with _domain_lock: