/
/
/
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 set_task_report(self, task_id: str, markdown: str | None) -> None:
493 """
494 Set the Markdown report for a task.
495
496 :param task_id: The id of the task to update.
497 :param markdown: Markdown report content or None to clear.
498 """
499 if get_ident() != self.mass.loop_thread_id:
500 self.mass.loop.call_soon_threadsafe(self.set_task_report, task_id, markdown)
501 return
502 if not (managed := self._tasks.get(task_id)):
503 return
504 self._update_task_report(managed, markdown)
505
506 def set_current_task_report(self, markdown: str | None) -> None:
507 """
508 Set the Markdown report for the task active in the current async context.
509
510 :param markdown: Markdown report content or None to clear.
511 """
512 if task_context := ACTIVE_TASK_CONTEXT.get():
513 task_context.set_report(markdown)
514
515 def add_task_failure(self, task_id: str, message: str) -> None:
516 """
517 Record a non-fatal failure for a task.
518
519 :param task_id: The id of the task to record the failure on.
520 :param message: Human-readable failure description.
521 """
522 if get_ident() != self.mass.loop_thread_id:
523 self.mass.loop.call_soon_threadsafe(self.add_task_failure, task_id, message)
524 return
525 if not (managed := self._tasks.get(task_id)):
526 return
527 task_info = managed.task_info
528 task_info.failure_count += 1
529 message = message.strip()
530 if message:
531 task_info.failure_messages.append(message)
532 if len(task_info.failure_messages) > DEFAULT_TASK_FAILURE_MESSAGES:
533 del task_info.failure_messages[
534 : len(task_info.failure_messages) - DEFAULT_TASK_FAILURE_MESSAGES
535 ]
536 task_info.updated_at = utcnow()
537 self._schedule_task_update()
538
539 def get_tasks_by_metadata(self, **metadata: TaskMetadataValue) -> list[BackgroundTask]:
540 """
541 Return tasks matching the given metadata key/value pairs.
542
543 :param metadata: Key/value pairs that must all match on a task's metadata.
544 """
545 result: list[BackgroundTask] = []
546 for managed in self._tasks.values():
547 if all(managed.task_info.metadata.get(key) == value for key, value in metadata.items()):
548 result.append(managed.task_info)
549 return result
550
551 def _unregister_task(
552 self, task_id: str, clear_persisted_state: bool = True
553 ) -> asyncio.Task[Any] | None:
554 """
555 Unregister a managed task and cancel any active work.
556
557 Returns the cancelled asyncio task if one was still running, so callers can wait
558 for it to unwind. Final bookkeeping for such a task happens in _finalize_task_run.
559 """
560 if not (managed := self._tasks.get(task_id)):
561 return None
562 managed.removed = True
563 managed.clear_persisted_state_on_remove = clear_persisted_state
564 self.mass.cancel_timer(get_task_timer_id(task_id))
565 self._remove_from_pending(task_id)
566 if managed.current_task and not managed.current_task.done():
567 managed.current_task.cancel()
568 return managed.current_task
569 self._tasks.pop(task_id, None)
570 if clear_persisted_state:
571 self._clear_scheduled_task_state(task_id)
572 self._schedule_task_update(force=True)
573 return None
574
575 def _get_managed_task(self, task_id: str) -> ManagedTask:
576 """Return runtime state for a managed task."""
577 if not (managed := self._tasks.get(task_id)):
578 raise InvalidDataError(f"Task {task_id} not found")
579 return managed
580
581 def _get_visible_managed_task(self, task_id: str, user: User | None) -> ManagedTask:
582 """Return a managed task if it is visible to the given user."""
583 managed = self._get_managed_task(task_id)
584 if (
585 user is not None
586 and not has_scope(user, Scope.SYSTEM_MANAGE)
587 and managed.task_info.user_id != user.user_id
588 ):
589 raise InvalidDataError(f"Task {task_id} not found")
590 return managed
591
592 def _append_task_log(self, task_id: str, line: str) -> None:
593 """Append a log line to a task."""
594 if get_ident() != self.mass.loop_thread_id:
595 self.mass.loop.call_soon_threadsafe(self._append_task_log, task_id, line)
596 return
597 if not (managed := self._tasks.get(task_id)):
598 return
599 logs = managed.task_info.logs
600 logs.append(line)
601 if len(logs) > managed.max_log_lines:
602 del logs[: len(logs) - managed.max_log_lines]
603 managed.task_info.updated_at = utcnow()
604 self._schedule_task_update()
605
606 def _set_task_report_for_run(
607 self,
608 task_id: str,
609 markdown: str | None,
610 *,
611 run_token: str,
612 ) -> None:
613 """Set a task report only while its originating run is active."""
614 if get_ident() != self.mass.loop_thread_id:
615 self.mass.loop.call_soon_threadsafe(
616 partial(
617 self._set_task_report_for_run,
618 task_id,
619 markdown,
620 run_token=run_token,
621 )
622 )
623 return
624 if not (managed := self._tasks.get(task_id)):
625 return
626 if managed.run_token != run_token or managed.task_info.status != TaskStatus.RUNNING:
627 return
628 self._update_task_report(managed, markdown)
629
630 def _update_task_report(self, managed: ManagedTask, markdown: str | None) -> None:
631 """Update a managed task report and notify task consumers."""
632 managed.task_info.report = markdown
633 managed.task_info.updated_at = utcnow()
634 self._persist_scheduled_task_state(managed)
635 self._schedule_task_update()
636
637 def _append_task_lifecycle_log(
638 self,
639 task_id: str,
640 *,
641 level: int,
642 message: str,
643 created_at: datetime | None = None,
644 ) -> None:
645 """Append a synthetic lifecycle log line using the default task log format."""
646 self._append_task_log(
647 task_id,
648 format_task_log_line(
649 message,
650 level=level,
651 logger_name=self.logger.name,
652 created_at=created_at,
653 ),
654 )
655
656 def _mark_task_running(self, managed: ManagedTask) -> None:
657 """Update task state for the start of a managed run."""
658 task_info = managed.task_info
659 task_info.status = TaskStatus.RUNNING
660 task_info.started_at = utcnow()
661 task_info.last_run = task_info.started_at
662 task_info.updated_at = task_info.started_at
663 self._persist_scheduled_task_state(managed)
664 self._append_task_lifecycle_log(
665 task_info.id,
666 level=logging.INFO,
667 message="Task started",
668 created_at=task_info.started_at,
669 )
670 self._schedule_task_update(force=True)
671
672 def _finalize_task_run(self, managed: ManagedTask) -> None:
673 """Finalize task bookkeeping after a managed run."""
674 task_info = managed.task_info
675 task_info.finished_at = utcnow()
676 task_info.updated_at = task_info.finished_at
677 managed.current_task = None
678 if managed.removed:
679 self._tasks.pop(task_info.id, None)
680 if managed.clear_persisted_state_on_remove:
681 self._clear_scheduled_task_state(task_info.id)
682 elif managed.is_scheduled:
683 self._schedule_managed_task(managed)
684 self._persist_scheduled_task_state(managed)
685 trim_finished_history(self._tasks, MAX_FINISHED_TASK_HISTORY)
686 self._schedule_task_update(force=True)
687 self._start_pending_tasks()
688
689 def _queue_task(
690 self,
691 managed: ManagedTask,
692 *,
693 reset_logs: bool,
694 run_user_id: str | None = None,
695 ) -> None:
696 """Queue a task for execution."""
697 if managed.removed:
698 raise InvalidDataError(f"Task {managed.task_info.id} is no longer available")
699 if managed.task_info.status in (TaskStatus.PENDING, TaskStatus.RUNNING):
700 return
701 self.mass.cancel_timer(get_task_timer_id(managed.task_info.id))
702 managed.run_token = uuid4().hex
703 if reset_logs:
704 managed.task_info.logs.clear()
705 managed.task_info.progress = None
706 managed.task_info.progress_text = None
707 managed.task_info.last_error = None
708 managed.task_info.failure_count = 0
709 managed.task_info.failure_messages.clear()
710 managed.task_info.report = None
711 managed.task_info.finished_at = None
712 managed.task_info.status = TaskStatus.PENDING
713 managed.task_info.last_run_user_id = run_user_id
714 managed.task_info.started_at = None
715 managed.task_info.next_run = None
716 managed.task_info.updated_at = utcnow()
717 self._persist_scheduled_task_state(managed)
718 if managed.task_info.id not in self._pending_task_ids:
719 if managed.priority:
720 self._pending_task_ids.appendleft(managed.task_info.id)
721 else:
722 self._pending_task_ids.append(managed.task_info.id)
723 self._schedule_task_update(force=True)
724 self._start_pending_tasks()
725
726 def _start_pending_tasks(self) -> None:
727 """Start queued tasks while concurrency allows it."""
728 while self._running_tasks_count < self._max_concurrent_tasks and self._pending_task_ids:
729 task_id = self._pending_task_ids.popleft()
730 if not (managed := self._tasks.get(task_id)) or managed.removed:
731 continue
732 if managed.task_info.status != TaskStatus.PENDING:
733 continue
734 managed.current_task = self.mass.create_task(self._run_task(managed))
735
736 @property
737 def _running_tasks_count(self) -> int:
738 """Return count of currently running managed tasks."""
739 return sum(
740 1 for managed in self._tasks.values() if managed.task_info.status == TaskStatus.RUNNING
741 )
742
743 async def _run_task(self, managed: ManagedTask) -> None:
744 """Run a managed task."""
745 task_info = managed.task_info
746 self._mark_task_running(managed)
747 task_context = TaskExecutionContext(
748 task_id=task_info.id,
749 get_task=self.get_task,
750 update_progress=self.update_task_progress,
751 update_progress_text=self.update_task_progress_text,
752 add_failure=self.add_task_failure,
753 update_report=partial(
754 self._set_task_report_for_run,
755 run_token=managed.run_token,
756 ),
757 )
758 token = ACTIVE_TASK_ID.set(task_info.id)
759 context_token = ACTIVE_TASK_CONTEXT.set(task_context)
760 try:
761 await managed.handler()
762 except asyncio.CancelledError:
763 task_info.status = TaskStatus.CANCELLED
764 task_info.last_error = None
765 now = utcnow()
766 self._append_task_lifecycle_log(
767 task_info.id,
768 level=logging.WARNING,
769 message="Task cancelled",
770 created_at=now,
771 )
772 except Exception as err:
773 task_info.status = TaskStatus.FAILED
774 task_info.last_error = str(err)
775 self.logger.warning(
776 "Background task %s failed: %s",
777 task_info.name,
778 str(err),
779 exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
780 )
781 now = utcnow()
782 self._append_task_lifecycle_log(
783 task_info.id,
784 level=logging.ERROR,
785 message=f"Task failed: {err}",
786 created_at=now,
787 )
788 else:
789 if task_info.failure_count:
790 now = utcnow()
791 task_info.status = TaskStatus.PARTIAL_SUCCESS
792 self._append_task_lifecycle_log(
793 task_info.id,
794 level=logging.WARNING,
795 message=f"Task completed with {task_info.failure_count} issue(s)",
796 created_at=now,
797 )
798 else:
799 task_info.status = TaskStatus.SUCCESS
800 now = utcnow()
801 self._append_task_lifecycle_log(
802 task_info.id,
803 level=logging.INFO,
804 message="Task completed successfully",
805 created_at=now,
806 )
807 finally:
808 ACTIVE_TASK_CONTEXT.reset(context_token)
809 ACTIVE_TASK_ID.reset(token)
810 self._finalize_task_run(managed)
811
812 def _cancel_managed_task(self, managed: ManagedTask) -> None:
813 """Cancel a pending or running task."""
814 task_info = managed.task_info
815 self.mass.cancel_timer(get_task_timer_id(task_info.id))
816 if managed.current_task and not managed.current_task.done():
817 managed.current_task.cancel()
818 elif task_info.status == TaskStatus.PENDING:
819 self._remove_from_pending(task_info.id)
820 task_info.status = TaskStatus.CANCELLED
821 task_info.finished_at = utcnow()
822 task_info.updated_at = task_info.finished_at
823 if managed.is_scheduled:
824 self._schedule_managed_task(managed)
825 elif managed.is_scheduled:
826 self._schedule_managed_task(managed)
827 self._persist_scheduled_task_state(managed)
828 self._schedule_task_update(force=True)
829
830 def _schedule_managed_task(self, managed: ManagedTask, delay: float | None = None) -> None:
831 """Schedule the next recurring execution of a task."""
832 if managed.removed or not managed.task_info.schedule:
833 return
834 if not managed.task_info.schedule.enabled:
835 managed.timer_delay = None
836 managed.task_info.next_run = None
837 managed.task_info.updated_at = utcnow()
838 self._persist_scheduled_task_state(managed)
839 self._schedule_task_update(force=True)
840 return
841 delay, next_run = get_task_schedule_delay(
842 managed.task_info.schedule,
843 last_run=managed.task_info.last_run,
844 fallback_delay=delay,
845 )
846 managed.timer_delay = delay
847 managed.task_info.next_run = next_run
848 managed.task_info.updated_at = utcnow()
849 self._persist_scheduled_task_state(managed)
850 self.mass.call_later(
851 delay,
852 partial(self._queue_task, managed, reset_logs=True),
853 task_id=get_task_timer_id(managed.task_info.id),
854 )
855 self._schedule_task_update(force=True)
856
857 def _restore_scheduled_task_state(self, task_info: BackgroundTask) -> None:
858 """Restore persisted runtime state for a scheduled task."""
859 if task_info.schedule is None:
860 return
861 states = self._get_persisted_task_states()
862 if not (state := states.get(task_info.id)) or not isinstance(state, dict):
863 return
864 restore_task_state(task_info, state)
865
866 def _persist_scheduled_task_state(self, managed: ManagedTask) -> None:
867 """Persist runtime state for a scheduled task."""
868 if not managed.is_scheduled:
869 return
870 updated_states = dict(self._get_persisted_task_states())
871 updated_states[managed.task_info.id] = serialize_task_state(managed.task_info)
872 self._set_persisted_task_states(updated_states)
873
874 def _clear_scheduled_task_state(self, task_id: str) -> None:
875 """Remove persisted runtime state for a scheduled task."""
876 updated_states = dict(self._get_persisted_task_states())
877 if task_id not in updated_states:
878 return
879 updated_states.pop(task_id, None)
880 self._set_persisted_task_states(updated_states)
881
882 def _get_persisted_task_states(self) -> dict[str, Any]:
883 """Return persisted runtime state for scheduled tasks."""
884 states = self.mass.config.get(f"core/{self.domain}/{TASK_STATE_CONFIG_KEY}", {})
885 return states if isinstance(states, dict) else {}
886
887 def _set_persisted_task_states(self, states: dict[str, Any]) -> None:
888 """Persist runtime state for scheduled tasks."""
889 # this state lives next to the config values rather than inside them, so the core
890 # config block has to be created up front. a bare set() would build the parent
891 # dicts on the fly and leave a block that no longer parses as a CoreConfig.
892 self.mass.config.ensure_core_config_base(self.domain)
893 self.mass.config.set(f"core/{self.domain}/{TASK_STATE_CONFIG_KEY}", states)
894
895 @staticmethod
896 def _resolve_schedule(
897 *,
898 schedule: TaskSchedule | None,
899 ) -> TaskSchedule:
900 """Resolve the requested schedule configuration."""
901 if schedule is None:
902 raise InvalidDataError("Scheduled task requires a schedule")
903 return schedule
904
905 @staticmethod
906 def _resolve_updated_schedule(
907 *,
908 current_schedule: TaskSchedule,
909 schedule: TaskSchedule | None,
910 ) -> TaskSchedule:
911 """Resolve an updated schedule while preserving enabled state."""
912 if schedule is None:
913 raise InvalidDataError("Updated schedule requires a schedule")
914 schedule.enabled = current_schedule.enabled
915 return schedule
916
917 @staticmethod
918 def _validate_progress(progress: int | None) -> int | None:
919 """Validate task progress percentage."""
920 if progress is None:
921 return None
922 if isinstance(progress, bool) or not isinstance(progress, int):
923 raise InvalidDataError("Task progress must be an integer percentage")
924 if not 0 <= progress <= 100:
925 raise InvalidDataError("Task progress must be between 0 and 100")
926 return progress
927
928 def _remove_from_pending(self, task_id: str) -> None:
929 """Remove a task from the pending queue."""
930 with suppress(ValueError):
931 self._pending_task_ids.remove(task_id)
932
933 def _schedule_task_update(self, *, force: bool = False) -> None:
934 """Coalesce task update events while keeping lifecycle updates responsive."""
935 if get_ident() != self.mass.loop_thread_id:
936 self.mass.loop.call_soon_threadsafe(partial(self._schedule_task_update, force=force))
937 return
938 now = self.mass.loop.time()
939 if force:
940 delay = TASK_LIFECYCLE_UPDATE_DEBOUNCE
941 else:
942 delay = max(
943 TASK_ACTIVITY_UPDATE_INTERVAL - (now - self._last_task_update_signal),
944 TASK_LIFECYCLE_UPDATE_DEBOUNCE,
945 )
946 scheduled_at = now + delay
947 if (
948 self._scheduled_task_update_at is not None
949 and self._scheduled_task_update_at <= scheduled_at
950 ):
951 return
952 self._scheduled_task_update_at = scheduled_at
953 self.mass.call_later(delay, self._signal_task_update, task_id=TASK_UPDATE_TIMER_ID)
954
955 def _signal_task_update(self) -> None:
956 """Emit the current managed task list."""
957 self._scheduled_task_update_at = None
958 self._last_task_update_signal = self.mass.loop.time()
959 self.mass.signal_event(EventType.TASKS_UPDATED, data=self.list_tasks_for_user(None))
960
961
962def _namespaced_translation_key(translation_key: str | None) -> str | None:
963 """
964 Namespace a bare task key under the shared ``background_task`` group.
965
966 Callers pass just the task key (e.g. ``database_cleanup``); the ``background_task`` group is
967 implicit for tasks and added here. A key that already carries a namespace (anything containing
968 a ``.``, e.g. a fully-qualified key) is returned unchanged.
969 """
970 if translation_key and "." not in translation_key:
971 return f"background_task.{translation_key}"
972 return translation_key
973