music-assistant-server

1"""Controller that manages long running background tasks."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections import deque
8from collections.abc import Awaitable, Callable
9from contextlib import suppress
10from datetime import datetime
11from functools import partial
12from threading import get_ident
13from typing import TYPE_CHECKING, Any, cast
14from uuid import uuid4
15
16from music_assistant_models.auth import Scope, User
17from music_assistant_models.background_task import (
18    BackgroundTask,
19    TaskMetadata,
20    TaskMetadataValue,
21    TaskSchedule,
22)
23from music_assistant_models.enums import EventType, TaskStatus
24from music_assistant_models.errors import InvalidDataError
25
26from music_assistant.constants import (
27    CONF_ENTRY_MAX_CONCURRENT_TASKS,
28    CONF_MAX_CONCURRENT_TASKS,
29)
30from music_assistant.controllers.webserver.helpers.auth_middleware import (
31    get_current_user,
32    has_scope,
33)
34from music_assistant.helpers.api import api_command
35from music_assistant.models.core_controller import CoreController
36
37from .constants import (
38    ACTIVE_TASK_ID,
39    DEFAULT_MAX_CONCURRENT_TASKS,
40    DEFAULT_TASK_FAILURE_MESSAGES,
41    DEFAULT_TASK_LOG_LINES,
42    MAX_FINISHED_TASK_HISTORY,
43    TASK_ACTIVITY_UPDATE_INTERVAL,
44    TASK_CANCEL_TIMEOUT,
45    TASK_LIFECYCLE_UPDATE_DEBOUNCE,
46    TASK_STATE_CONFIG_KEY,
47    TASK_UPDATE_TIMER_ID,
48)
49from .context import ACTIVE_TASK_CONTEXT, TaskExecutionContext
50from .helpers import (
51    TaskLogHandler,
52    format_task_log_line,
53    get_task_schedule_delay,
54    get_task_timer_id,
55    get_visible_tasks,
56    merge_task_schedule_state,
57    restore_task_state,
58    serialize_task_schedule_state,
59    serialize_task_state,
60    trim_finished_history,
61    utcnow,
62)
63from .models import ManagedTask
64
65if TYPE_CHECKING:
66    from music_assistant_models.config_entries import (
67        ConfigEntry,
68        CoreConfig,
69    )
70
71    from music_assistant import MusicAssistant
72    from music_assistant.helpers.json import SerializableType
73
74
75class TasksController(CoreController):
76    """Controller that manages long running background tasks."""
77
78    domain = "tasks"
79
80    def __init__(self, mass: MusicAssistant) -> None:
81        """Initialize controller."""
82        super().__init__(mass)
83        self.manifest.name = "Background tasks"
84        self.manifest.description = "Manage long running scheduled, user and system tasks."
85        self.manifest.icon = "playlist-play"
86        self._tasks: dict[str, ManagedTask] = {}
87        self._pending_task_ids: deque[str] = deque()
88        self._log_handler: TaskLogHandler | None = None
89        self._max_concurrent_tasks = DEFAULT_MAX_CONCURRENT_TASKS
90        self._last_task_update_signal = 0.0
91        self._scheduled_task_update_at: float | None = None
92
93    async def setup(self, config: CoreConfig) -> None:
94        """Set up the controller."""
95        self.config = config
96        self._max_concurrent_tasks = cast(
97            "int", config.get_value(CONF_MAX_CONCURRENT_TASKS, DEFAULT_MAX_CONCURRENT_TASKS)
98        )
99        if self._log_handler is None:
100            self._log_handler = TaskLogHandler(self.mass, self._append_task_log)
101            logging.getLogger().addHandler(self._log_handler)
102
103    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
104        """Return all Config Entries for this core module (if any)."""
105        return (CONF_ENTRY_MAX_CONCURRENT_TASKS,)
106
107    async def close(self) -> None:
108        """Clean up the controller."""
109        for task_id in list(self._tasks):
110            self._unregister_task(task_id, clear_persisted_state=False)
111        if self._log_handler is not None:
112            logging.getLogger().removeHandler(self._log_handler)
113            self._log_handler = None
114
115    async def get_diagnostics(self) -> dict[str, SerializableType]:
116        """Return diagnostics info for this controller to include in diagnostics reports."""
117        by_status: dict[str, int] = {}
118        for managed in self._tasks.values():
119            status = managed.task_info.status.value
120            by_status[status] = by_status.get(status, 0) + 1
121        return {
122            "total": len(self._tasks),
123            "by_status": by_status,
124            "scheduled": sum(managed.is_scheduled for managed in self._tasks.values()),
125            "pending_queue": len(self._pending_task_ids),
126            "max_concurrent": self._max_concurrent_tasks,
127        }
128
129    @api_command("tasks/list", required_scope=Scope.SYSTEM_READ)
130    def list_tasks(self) -> list[BackgroundTask]:
131        """Return all visible managed tasks."""
132        return self.list_tasks_for_user(get_current_user())
133
134    def list_tasks_for_user(self, user: User | None) -> list[BackgroundTask]:
135        """Return tasks visible to the given user."""
136        return [managed.task_info for managed in get_visible_tasks(self._tasks.values(), user)]
137
138    @api_command("tasks/get", required_scope=Scope.SYSTEM_READ)
139    def get_task(self, task_id: str) -> BackgroundTask:
140        """Return a single task by id."""
141        return self._get_visible_managed_task(task_id, get_current_user()).task_info
142
143    @api_command("tasks/log", required_scope=Scope.SYSTEM_READ)
144    def get_task_log(self, task_id: str) -> str:
145        """Return the log buffer for a single task."""
146        return "\n".join(self._get_visible_managed_task(task_id, get_current_user()).task_info.logs)
147
148    @api_command("tasks/run", required_scope=Scope.SYSTEM_MANAGE)
149    def run_task(self, task_id: str) -> BackgroundTask:
150        """Queue a task for immediate execution."""
151        managed = self._get_managed_task(task_id)
152        if not managed.is_scheduled:
153            raise InvalidDataError(f"Task {task_id} can not be run manually")
154        user = get_current_user()
155        self._queue_task(
156            managed,
157            reset_logs=True,
158            run_user_id=user.user_id if user else None,
159        )
160        return managed.task_info
161
162    @api_command("tasks/retry", required_scope=Scope.SYSTEM_MANAGE)
163    def retry_task(self, task_id: str) -> BackgroundTask:
164        """Retry a failed or cancelled task."""
165        managed = self._get_managed_task(task_id)
166        if not managed.task_info.allow_retry:
167            raise InvalidDataError(f"Task {task_id} can not be retried")
168        if managed.task_info.status not in (
169            TaskStatus.FAILED,
170            TaskStatus.CANCELLED,
171            TaskStatus.PARTIAL_SUCCESS,
172        ):
173            raise InvalidDataError(f"Task {task_id} is not in a retryable state")
174        user = get_current_user()
175        self._queue_task(
176            managed,
177            reset_logs=True,
178            run_user_id=user.user_id if user else None,
179        )
180        return managed.task_info
181
182    @api_command("tasks/cancel", required_scope=Scope.SYSTEM_MANAGE)
183    def cancel_task(self, task_id: str) -> BackgroundTask:
184        """Cancel a pending or running task."""
185        managed = self._get_managed_task(task_id)
186        self._cancel_managed_task(managed)
187        return managed.task_info
188
189    @api_command("tasks/set_enabled", required_scope=Scope.SYSTEM_MANAGE)
190    def set_task_enabled(self, task_id: str, enabled: bool) -> BackgroundTask:
191        """Enable or disable automatic scheduling for a recurring task."""
192        managed = self._get_managed_task(task_id)
193        if not managed.task_info.schedule:
194            raise InvalidDataError(f"Task {task_id} does not have a recurring schedule")
195        managed.task_info.schedule.enabled = enabled
196        self.mass.cancel_timer(get_task_timer_id(task_id))
197        if enabled:
198            if not managed.is_active:
199                self._schedule_managed_task(managed)
200            else:
201                managed.task_info.updated_at = utcnow()
202                self._persist_scheduled_task_state(managed)
203                self._schedule_task_update(force=True)
204        else:
205            managed.timer_delay = None
206            managed.task_info.next_run = None
207            managed.task_info.updated_at = utcnow()
208            self._persist_scheduled_task_state(managed)
209            self._schedule_task_update(force=True)
210        return managed.task_info
211
212    @api_command("tasks/update_schedule", required_scope=Scope.SYSTEM_MANAGE)
213    def update_task_schedule(
214        self,
215        task_id: str,
216        schedule: TaskSchedule,
217    ) -> BackgroundTask:
218        """Update the schedule definition for a recurring task."""
219        managed = self._get_managed_task(task_id)
220        current_schedule = managed.task_info.schedule
221        if current_schedule is None:
222            raise InvalidDataError(f"Task {task_id} does not have a recurring schedule")
223        managed.task_info.schedule = self._resolve_updated_schedule(
224            current_schedule=current_schedule,
225            schedule=schedule,
226        )
227        self.mass.cancel_timer(get_task_timer_id(task_id))
228        managed.task_info.next_run = None
229        managed.task_info.updated_at = utcnow()
230        if not managed.is_active:
231            self._schedule_managed_task(managed)
232        else:
233            self._persist_scheduled_task_state(managed)
234            self._schedule_task_update(force=True)
235        return managed.task_info
236
237    @api_command("tasks/remove", required_scope=Scope.SYSTEM_MANAGE)
238    def remove_task(self, task_id: str) -> None:
239        """Remove a finished task from history."""
240        managed = self._get_managed_task(task_id)
241        if not managed.can_remove:
242            raise InvalidDataError(f"Task {task_id} can not be removed")
243        self._tasks.pop(task_id, None)
244        self._schedule_task_update(force=True)
245
246    @api_command("tasks/clear_finished", required_scope=Scope.SYSTEM_MANAGE)
247    def clear_finished_tasks(self) -> None:
248        """Remove finished non-scheduled tasks from history."""
249        for task_id in [task_id for task_id, task in self._tasks.items() if task.can_remove]:
250            self._tasks.pop(task_id, None)
251        self._schedule_task_update(force=True)
252
253    def run_background_task(  # noqa: PLR0913
254        self,
255        *,
256        name: str,
257        handler: Callable[[], Awaitable[Any]],
258        task_id: str | None = None,
259        translation_key: str | None = None,
260        translation_args: list[Any] | None = None,
261        translation_owner: str | None = None,
262        user_id: str | None = None,
263        metadata: TaskMetadata | None = None,
264        allow_retry: bool = False,
265        allow_cancel: bool = True,
266        priority: bool = False,
267        max_log_lines: int = DEFAULT_TASK_LOG_LINES,
268    ) -> BackgroundTask:
269        """
270        Create and queue a long running background task.
271
272        :param name: Human-readable display name for the task.
273        :param handler: Async callable that performs the actual work.
274        :param task_id: Optional deterministic id. Auto-generated if not provided.
275            When a task with the same id already exists and is active,
276            the existing task is returned as-is. If inactive, it is replaced.
277        :param translation_key: Optional translation key for localised task names.
278        :param translation_args: Optional arguments for the translation key.
279        :param translation_owner: Owner namespace the (relative) translation_key resolves under,
280            e.g. the calling module's ``translation_owner`` ("core.<domain>"/"provider.<domain>").
281        :param user_id: Optional user id that initiated the task.
282        :param metadata: Optional key/value metadata attached to the task.
283        :param allow_retry: Whether the task can be retried after failure.
284        :param allow_cancel: Whether the task can be cancelled by a user.
285        :param priority: When True, the task is inserted at the front of the pending queue
286            so it runs before lower-priority tasks. Use this for user-initiated actions that
287            should not be delayed by background work such as metadata refreshes.
288        :param max_log_lines: Maximum number of log lines to retain for this task.
289        """
290        resolved_task_id = task_id or uuid4().hex
291        if existing := self._tasks.get(resolved_task_id):
292            if existing.is_active:
293                return existing.task_info
294            self._tasks.pop(resolved_task_id, None)
295
296        task_info = BackgroundTask(
297            id=resolved_task_id,
298            name=name,
299            status=TaskStatus.IDLE,
300            translation_key=_namespaced_translation_key(translation_key),
301            translation_args=translation_args or [],
302            translation_owner=translation_owner,
303            user_id=user_id,
304            metadata=metadata or {},
305            allow_retry=allow_retry,
306            allow_cancel=allow_cancel,
307        )
308        managed = ManagedTask(
309            task_info=task_info,
310            handler=handler,
311            priority=priority,
312            max_log_lines=max_log_lines,
313        )
314        self._tasks[task_info.id] = managed
315        self._queue_task(managed, reset_logs=True, run_user_id=user_id)
316        return task_info
317
318    def register_scheduled_task(  # noqa: PLR0913
319        self,
320        *,
321        task_id: str,
322        name: str,
323        handler: Callable[[], Awaitable[Any]],
324        schedule: TaskSchedule,
325        initial_delay: float | None = None,
326        translation_key: str | None = None,
327        translation_args: list[Any] | None = None,
328        translation_owner: str | None = None,
329        metadata: TaskMetadata | None = None,
330        allow_retry: bool = False,
331        allow_cancel: bool = True,
332    ) -> BackgroundTask:
333        """
334        Register or update a recurring scheduled task.
335
336        :param task_id: Deterministic id for the scheduled task.
337        :param name: Human-readable display name for the task.
338        :param handler: Async callable that performs the actual work.
339        :param schedule: Schedule definition controlling when the task runs.
340        :param initial_delay: Optional delay in seconds before the first run.
341        :param translation_key: Optional translation key for localised task names.
342        :param translation_args: Optional arguments for the translation key.
343        :param translation_owner: Owner namespace the (relative) translation_key resolves under,
344            e.g. the calling module's ``translation_owner`` ("core.<domain>"/"provider.<domain>").
345        :param metadata: Optional key/value metadata attached to the task.
346        :param allow_retry: Whether the task can be retried after failure.
347        :param allow_cancel: Whether the task can be cancelled by a user.
348        """
349        resolved_schedule = self._resolve_schedule(schedule=schedule)
350        if existing := self._tasks.get(task_id):
351            task_info = existing.task_info
352            task_info.name = name
353            task_info.translation_key = _namespaced_translation_key(translation_key)
354            task_info.translation_args = translation_args or []
355            task_info.translation_owner = translation_owner
356            task_info.metadata = metadata or {}
357            if task_info.schedule is not None:
358                task_info.schedule = merge_task_schedule_state(
359                    resolved_schedule,
360                    serialize_task_schedule_state(task_info.schedule) or {},
361                )
362            else:
363                task_info.schedule = resolved_schedule
364            task_info.allow_retry = allow_retry
365            task_info.allow_cancel = allow_cancel
366            task_info.updated_at = utcnow()
367            existing.handler = handler
368            existing.removed = False
369            if not existing.is_active:
370                self._schedule_managed_task(existing, initial_delay)
371            self._persist_scheduled_task_state(existing)
372            self._schedule_task_update(force=True)
373            return task_info
374
375        task_info = BackgroundTask(
376            id=task_id,
377            name=name,
378            status=TaskStatus.IDLE,
379            translation_key=_namespaced_translation_key(translation_key),
380            translation_args=translation_args or [],
381            translation_owner=translation_owner,
382            metadata=metadata or {},
383            schedule=resolved_schedule,
384            allow_retry=allow_retry,
385            allow_cancel=allow_cancel,
386        )
387        self._restore_scheduled_task_state(task_info)
388        managed = ManagedTask(task_info=task_info, handler=handler)
389        self._tasks[task_id] = managed
390        self._schedule_managed_task(managed, initial_delay)
391        self._persist_scheduled_task_state(managed)
392        self._schedule_task_update(force=True)
393        return task_info
394
395    def unregister_scheduled_task(self, task_id: str, clear_persisted_state: bool = True) -> None:
396        """
397        Unregister a recurring scheduled task and cancel any active work.
398
399        If a stale ad-hoc task exists with the same deterministic task id,
400        remove that too so provider/task re-registration can recover cleanly.
401
402        :param task_id: The id of the scheduled task to unregister.
403        :param clear_persisted_state: Whether to remove persisted state from config.
404        """
405        self._unregister_task(task_id, clear_persisted_state)
406
407    async def unregister_scheduled_task_and_wait(
408        self,
409        task_id: str,
410        clear_persisted_state: bool = True,
411        timeout: float = TASK_CANCEL_TIMEOUT,
412    ) -> bool:
413        """
414        Unregister a recurring scheduled task and wait for a running task to unwind.
415
416        Cancellation is only a request: with plain ``unregister_scheduled_task`` the task
417        can still be running (unwinding) when the caller continues. Use this variant from
418        teardown paths that destroy state the task may still be touching, such as unloading
419        a provider.
420
421        Note that a task blocked in a thread (``asyncio.to_thread`` and friends) unwinds
422        immediately while its thread keeps running, so this does not guarantee that all
423        work of the task has stopped.
424
425        :param task_id: The id of the scheduled task to unregister.
426        :param clear_persisted_state: Whether to remove persisted state from config.
427        :param timeout: Maximum number of seconds to wait for the task to unwind.
428        :return: True if no task was left running, False if the wait timed out.
429        """
430        cancelled_task = self._unregister_task(task_id, clear_persisted_state)
431        if cancelled_task is None:
432            return True
433        if cancelled_task is asyncio.current_task():
434            # the task is cancelling itself (e.g. a sync task that triggers a provider
435            # unload); asyncio refuses a task awaiting itself so just let it unwind
436            return True
437        done, _pending = await asyncio.wait({cancelled_task}, timeout=timeout)
438        if not done:
439            self.logger.warning(
440                "Timeout while waiting for task %s to stop after cancellation", task_id
441            )
442            return False
443        return True
444
445    def update_task_progress(
446        self, task_id: str, progress: int | None, text: str | None = None
447    ) -> None:
448        """
449        Update progress for a task.
450
451        :param task_id: The id of the task to update.
452        :param progress: Progress percentage (0-100) or None to clear.
453        :param text: Optional progress description text.
454        """
455        if get_ident() != self.mass.loop_thread_id:
456            self.mass.loop.call_soon_threadsafe(self.update_task_progress, task_id, progress, text)
457            return
458        if not (managed := self._tasks.get(task_id)):
459            return
460        managed.task_info.progress = self._validate_progress(progress)
461        managed.task_info.progress_text = text
462        managed.task_info.updated_at = utcnow()
463        self._schedule_task_update()
464
465    def update_task_progress_text(self, task_id: str, text: str | None) -> None:
466        """
467        Update progress text for a task without changing the percentage.
468
469        :param task_id: The id of the task to update.
470        :param text: Progress description text or None to clear.
471        """
472        if get_ident() != self.mass.loop_thread_id:
473            self.mass.loop.call_soon_threadsafe(self.update_task_progress_text, task_id, text)
474            return
475        if not (managed := self._tasks.get(task_id)):
476            return
477        managed.task_info.progress_text = text
478        managed.task_info.updated_at = utcnow()
479        self._schedule_task_update()
480
481    def update_current_task_progress(self, progress: int | None, text: str | None = None) -> None:
482        """
483        Update progress for the task active in the current async context.
484
485        :param progress: Progress percentage (0-100) or None to clear.
486        :param text: Optional progress description text.
487        """
488        if not (task_id := ACTIVE_TASK_ID.get()):
489            return
490        self.update_task_progress(task_id, progress, text)
491
492    def add_task_failure(self, task_id: str, message: str) -> None:
493        """
494        Record a non-fatal failure for a task.
495
496        :param task_id: The id of the task to record the failure on.
497        :param message: Human-readable failure description.
498        """
499        if get_ident() != self.mass.loop_thread_id:
500            self.mass.loop.call_soon_threadsafe(self.add_task_failure, task_id, message)
501            return
502        if not (managed := self._tasks.get(task_id)):
503            return
504        task_info = managed.task_info
505        task_info.failure_count += 1
506        message = message.strip()
507        if message:
508            task_info.failure_messages.append(message)
509            if len(task_info.failure_messages) > DEFAULT_TASK_FAILURE_MESSAGES:
510                del task_info.failure_messages[
511                    : len(task_info.failure_messages) - DEFAULT_TASK_FAILURE_MESSAGES
512                ]
513        task_info.updated_at = utcnow()
514        self._schedule_task_update()
515
516    def get_tasks_by_metadata(self, **metadata: TaskMetadataValue) -> list[BackgroundTask]:
517        """
518        Return tasks matching the given metadata key/value pairs.
519
520        :param metadata: Key/value pairs that must all match on a task's metadata.
521        """
522        result: list[BackgroundTask] = []
523        for managed in self._tasks.values():
524            if all(managed.task_info.metadata.get(key) == value for key, value in metadata.items()):
525                result.append(managed.task_info)
526        return result
527
528    def _unregister_task(
529        self, task_id: str, clear_persisted_state: bool = True
530    ) -> asyncio.Task[Any] | None:
531        """
532        Unregister a managed task and cancel any active work.
533
534        Returns the cancelled asyncio task if one was still running, so callers can wait
535        for it to unwind. Final bookkeeping for such a task happens in _finalize_task_run.
536        """
537        if not (managed := self._tasks.get(task_id)):
538            return None
539        managed.removed = True
540        managed.clear_persisted_state_on_remove = clear_persisted_state
541        self.mass.cancel_timer(get_task_timer_id(task_id))
542        self._remove_from_pending(task_id)
543        if managed.current_task and not managed.current_task.done():
544            managed.current_task.cancel()
545            return managed.current_task
546        self._tasks.pop(task_id, None)
547        if clear_persisted_state:
548            self._clear_scheduled_task_state(task_id)
549        self._schedule_task_update(force=True)
550        return None
551
552    def _get_managed_task(self, task_id: str) -> ManagedTask:
553        """Return runtime state for a managed task."""
554        if not (managed := self._tasks.get(task_id)):
555            raise InvalidDataError(f"Task {task_id} not found")
556        return managed
557
558    def _get_visible_managed_task(self, task_id: str, user: User | None) -> ManagedTask:
559        """Return a managed task if it is visible to the given user."""
560        managed = self._get_managed_task(task_id)
561        if (
562            user is not None
563            and not has_scope(user, Scope.SYSTEM_MANAGE)
564            and managed.task_info.user_id != user.user_id
565        ):
566            raise InvalidDataError(f"Task {task_id} not found")
567        return managed
568
569    def _append_task_log(self, task_id: str, line: str) -> None:
570        """Append a log line to a task."""
571        if get_ident() != self.mass.loop_thread_id:
572            self.mass.loop.call_soon_threadsafe(self._append_task_log, task_id, line)
573            return
574        if not (managed := self._tasks.get(task_id)):
575            return
576        logs = managed.task_info.logs
577        logs.append(line)
578        if len(logs) > managed.max_log_lines:
579            del logs[: len(logs) - managed.max_log_lines]
580        managed.task_info.updated_at = utcnow()
581        self._schedule_task_update()
582
583    def _append_task_lifecycle_log(
584        self,
585        task_id: str,
586        *,
587        level: int,
588        message: str,
589        created_at: datetime | None = None,
590    ) -> None:
591        """Append a synthetic lifecycle log line using the default task log format."""
592        self._append_task_log(
593            task_id,
594            format_task_log_line(
595                message,
596                level=level,
597                logger_name=self.logger.name,
598                created_at=created_at,
599            ),
600        )
601
602    def _mark_task_running(self, managed: ManagedTask) -> None:
603        """Update task state for the start of a managed run."""
604        task_info = managed.task_info
605        task_info.status = TaskStatus.RUNNING
606        task_info.started_at = utcnow()
607        task_info.last_run = task_info.started_at
608        task_info.updated_at = task_info.started_at
609        self._persist_scheduled_task_state(managed)
610        self._append_task_lifecycle_log(
611            task_info.id,
612            level=logging.INFO,
613            message="Task started",
614            created_at=task_info.started_at,
615        )
616        self._schedule_task_update(force=True)
617
618    def _finalize_task_run(self, managed: ManagedTask) -> None:
619        """Finalize task bookkeeping after a managed run."""
620        task_info = managed.task_info
621        task_info.finished_at = utcnow()
622        task_info.updated_at = task_info.finished_at
623        managed.current_task = None
624        if managed.removed:
625            self._tasks.pop(task_info.id, None)
626            if managed.clear_persisted_state_on_remove:
627                self._clear_scheduled_task_state(task_info.id)
628        elif managed.is_scheduled:
629            self._schedule_managed_task(managed)
630            self._persist_scheduled_task_state(managed)
631        trim_finished_history(self._tasks, MAX_FINISHED_TASK_HISTORY)
632        self._schedule_task_update(force=True)
633        self._start_pending_tasks()
634
635    def _queue_task(
636        self,
637        managed: ManagedTask,
638        *,
639        reset_logs: bool,
640        run_user_id: str | None = None,
641    ) -> None:
642        """Queue a task for execution."""
643        if managed.removed:
644            raise InvalidDataError(f"Task {managed.task_info.id} is no longer available")
645        if managed.task_info.status in (TaskStatus.PENDING, TaskStatus.RUNNING):
646            return
647        self.mass.cancel_timer(get_task_timer_id(managed.task_info.id))
648        if reset_logs:
649            managed.task_info.logs.clear()
650            managed.task_info.progress = None
651            managed.task_info.progress_text = None
652            managed.task_info.last_error = None
653            managed.task_info.failure_count = 0
654            managed.task_info.failure_messages.clear()
655            managed.task_info.finished_at = None
656        managed.task_info.status = TaskStatus.PENDING
657        managed.task_info.last_run_user_id = run_user_id
658        managed.task_info.started_at = None
659        managed.task_info.next_run = None
660        managed.task_info.updated_at = utcnow()
661        if managed.task_info.id not in self._pending_task_ids:
662            if managed.priority:
663                self._pending_task_ids.appendleft(managed.task_info.id)
664            else:
665                self._pending_task_ids.append(managed.task_info.id)
666        self._schedule_task_update(force=True)
667        self._start_pending_tasks()
668
669    def _start_pending_tasks(self) -> None:
670        """Start queued tasks while concurrency allows it."""
671        while self._running_tasks_count < self._max_concurrent_tasks and self._pending_task_ids:
672            task_id = self._pending_task_ids.popleft()
673            if not (managed := self._tasks.get(task_id)) or managed.removed:
674                continue
675            if managed.task_info.status != TaskStatus.PENDING:
676                continue
677            managed.current_task = self.mass.create_task(self._run_task(managed))
678
679    @property
680    def _running_tasks_count(self) -> int:
681        """Return count of currently running managed tasks."""
682        return sum(
683            1 for managed in self._tasks.values() if managed.task_info.status == TaskStatus.RUNNING
684        )
685
686    async def _run_task(self, managed: ManagedTask) -> None:
687        """Run a managed task."""
688        task_info = managed.task_info
689        self._mark_task_running(managed)
690        task_context = TaskExecutionContext(
691            task_id=task_info.id,
692            get_task=self.get_task,
693            update_progress=self.update_task_progress,
694            update_progress_text=self.update_task_progress_text,
695            add_failure=self.add_task_failure,
696        )
697        token = ACTIVE_TASK_ID.set(task_info.id)
698        context_token = ACTIVE_TASK_CONTEXT.set(task_context)
699        try:
700            await managed.handler()
701        except asyncio.CancelledError:
702            task_info.status = TaskStatus.CANCELLED
703            task_info.last_error = None
704            now = utcnow()
705            self._append_task_lifecycle_log(
706                task_info.id,
707                level=logging.WARNING,
708                message="Task cancelled",
709                created_at=now,
710            )
711        except Exception as err:
712            task_info.status = TaskStatus.FAILED
713            task_info.last_error = str(err)
714            self.logger.warning(
715                "Background task %s failed: %s",
716                task_info.name,
717                str(err),
718                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
719            )
720            now = utcnow()
721            self._append_task_lifecycle_log(
722                task_info.id,
723                level=logging.ERROR,
724                message=f"Task failed: {err}",
725                created_at=now,
726            )
727        else:
728            if task_info.failure_count:
729                now = utcnow()
730                task_info.status = TaskStatus.PARTIAL_SUCCESS
731                self._append_task_lifecycle_log(
732                    task_info.id,
733                    level=logging.WARNING,
734                    message=f"Task completed with {task_info.failure_count} issue(s)",
735                    created_at=now,
736                )
737            else:
738                task_info.status = TaskStatus.SUCCESS
739                now = utcnow()
740                self._append_task_lifecycle_log(
741                    task_info.id,
742                    level=logging.INFO,
743                    message="Task completed successfully",
744                    created_at=now,
745                )
746        finally:
747            ACTIVE_TASK_CONTEXT.reset(context_token)
748            ACTIVE_TASK_ID.reset(token)
749            self._finalize_task_run(managed)
750
751    def _cancel_managed_task(self, managed: ManagedTask) -> None:
752        """Cancel a pending or running task."""
753        task_info = managed.task_info
754        self.mass.cancel_timer(get_task_timer_id(task_info.id))
755        if managed.current_task and not managed.current_task.done():
756            managed.current_task.cancel()
757        elif task_info.status == TaskStatus.PENDING:
758            self._remove_from_pending(task_info.id)
759            task_info.status = TaskStatus.CANCELLED
760            task_info.finished_at = utcnow()
761            task_info.updated_at = task_info.finished_at
762            if managed.is_scheduled:
763                self._schedule_managed_task(managed)
764        elif managed.is_scheduled:
765            self._schedule_managed_task(managed)
766        self._persist_scheduled_task_state(managed)
767        self._schedule_task_update(force=True)
768
769    def _schedule_managed_task(self, managed: ManagedTask, delay: float | None = None) -> None:
770        """Schedule the next recurring execution of a task."""
771        if managed.removed or not managed.task_info.schedule:
772            return
773        if not managed.task_info.schedule.enabled:
774            managed.timer_delay = None
775            managed.task_info.next_run = None
776            managed.task_info.updated_at = utcnow()
777            self._persist_scheduled_task_state(managed)
778            self._schedule_task_update(force=True)
779            return
780        delay, next_run = get_task_schedule_delay(
781            managed.task_info.schedule,
782            last_run=managed.task_info.last_run,
783            fallback_delay=delay,
784        )
785        managed.timer_delay = delay
786        managed.task_info.next_run = next_run
787        managed.task_info.updated_at = utcnow()
788        self._persist_scheduled_task_state(managed)
789        self.mass.call_later(
790            delay,
791            partial(self._queue_task, managed, reset_logs=True),
792            task_id=get_task_timer_id(managed.task_info.id),
793        )
794        self._schedule_task_update(force=True)
795
796    def _restore_scheduled_task_state(self, task_info: BackgroundTask) -> None:
797        """Restore persisted runtime state for a scheduled task."""
798        if task_info.schedule is None:
799            return
800        states = self._get_persisted_task_states()
801        if not (state := states.get(task_info.id)) or not isinstance(state, dict):
802            return
803        restore_task_state(task_info, state)
804
805    def _persist_scheduled_task_state(self, managed: ManagedTask) -> None:
806        """Persist runtime state for a scheduled task."""
807        if not managed.is_scheduled:
808            return
809        updated_states = dict(self._get_persisted_task_states())
810        updated_states[managed.task_info.id] = serialize_task_state(managed.task_info)
811        self._set_persisted_task_states(updated_states)
812
813    def _clear_scheduled_task_state(self, task_id: str) -> None:
814        """Remove persisted runtime state for a scheduled task."""
815        updated_states = dict(self._get_persisted_task_states())
816        if task_id not in updated_states:
817            return
818        updated_states.pop(task_id, None)
819        self._set_persisted_task_states(updated_states)
820
821    def _get_persisted_task_states(self) -> dict[str, Any]:
822        """Return persisted runtime state for scheduled tasks."""
823        states = self.mass.config.get(f"core/{self.domain}/{TASK_STATE_CONFIG_KEY}", {})
824        return states if isinstance(states, dict) else {}
825
826    def _set_persisted_task_states(self, states: dict[str, Any]) -> None:
827        """Persist runtime state for scheduled tasks."""
828        self.mass.config.set(f"core/{self.domain}/{TASK_STATE_CONFIG_KEY}", states)
829
830    @staticmethod
831    def _resolve_schedule(
832        *,
833        schedule: TaskSchedule | None,
834    ) -> TaskSchedule:
835        """Resolve the requested schedule configuration."""
836        if schedule is None:
837            raise InvalidDataError("Scheduled task requires a schedule")
838        return schedule
839
840    @staticmethod
841    def _resolve_updated_schedule(
842        *,
843        current_schedule: TaskSchedule,
844        schedule: TaskSchedule | None,
845    ) -> TaskSchedule:
846        """Resolve an updated schedule while preserving enabled state."""
847        if schedule is None:
848            raise InvalidDataError("Updated schedule requires a schedule")
849        schedule.enabled = current_schedule.enabled
850        return schedule
851
852    @staticmethod
853    def _validate_progress(progress: int | None) -> int | None:
854        """Validate task progress percentage."""
855        if progress is None:
856            return None
857        if isinstance(progress, bool) or not isinstance(progress, int):
858            raise InvalidDataError("Task progress must be an integer percentage")
859        if not 0 <= progress <= 100:
860            raise InvalidDataError("Task progress must be between 0 and 100")
861        return progress
862
863    def _remove_from_pending(self, task_id: str) -> None:
864        """Remove a task from the pending queue."""
865        with suppress(ValueError):
866            self._pending_task_ids.remove(task_id)
867
868    def _schedule_task_update(self, *, force: bool = False) -> None:
869        """Coalesce task update events while keeping lifecycle updates responsive."""
870        if get_ident() != self.mass.loop_thread_id:
871            self.mass.loop.call_soon_threadsafe(partial(self._schedule_task_update, force=force))
872            return
873        now = self.mass.loop.time()
874        if force:
875            delay = TASK_LIFECYCLE_UPDATE_DEBOUNCE
876        else:
877            delay = max(
878                TASK_ACTIVITY_UPDATE_INTERVAL - (now - self._last_task_update_signal),
879                TASK_LIFECYCLE_UPDATE_DEBOUNCE,
880            )
881        scheduled_at = now + delay
882        if (
883            self._scheduled_task_update_at is not None
884            and self._scheduled_task_update_at <= scheduled_at
885        ):
886            return
887        self._scheduled_task_update_at = scheduled_at
888        self.mass.call_later(delay, self._signal_task_update, task_id=TASK_UPDATE_TIMER_ID)
889
890    def _signal_task_update(self) -> None:
891        """Emit the current managed task list."""
892        self._scheduled_task_update_at = None
893        self._last_task_update_signal = self.mass.loop.time()
894        self.mass.signal_event(EventType.TASKS_UPDATED, data=self.list_tasks_for_user(None))
895
896
897def _namespaced_translation_key(translation_key: str | None) -> str | None:
898    """
899    Namespace a bare task key under the shared ``background_task`` group.
900
901    Callers pass just the task key (e.g. ``database_cleanup``); the ``background_task`` group is
902    implicit for tasks and added here. A key that already carries a namespace (anything containing
903    a ``.``, e.g. a fully-qualified key) is returned unchanged.
904    """
905    if translation_key and "." not in translation_key:
906        return f"background_task.{translation_key}"
907    return translation_key
908