music-assistant-server

1"""
2FastMCP sub-server for debug / troubleshooting tools.
3
4Spec: ``specs/inprogress/0005-debug-namespace.md``.
5
6All tools in this module are gated by off-by-default ConfigEntries
7(see ``provider/config.py``). With no tag enabled the entire namespace
8is invisible to MCP clients via ``TagFilterMiddleware``.
9"""
10# ruff: noqa: TID252  -- relative imports are the canonical MA-provider pattern.
11
12from __future__ import annotations
13
14import asyncio
15import importlib.metadata
16import logging
17import time
18from typing import TYPE_CHECKING, Any
19
20from fastmcp import Context, FastMCP
21from fastmcp.exceptions import ToolError
22from mcp.types import ToolAnnotations
23
24from ..debug.event_buffer import EventBuffer
25from ..debug.inspect_serializer import dump
26from ..debug.log_reader import SafeLogTail
27from ..models import (
28    ConfigValueDump,
29    EventBufferStats,
30    EventSnapshot,
31    HealthSummary,
32    LogStatsResult,
33    LogTailResult,
34    PackageVersions,
35    PlayerInspect,
36    ProviderConfigDump,
37    ProviderInspect,
38    ProviderList,
39    ProviderSummary,
40    QueueInspect,
41    ReloadResult,
42    RouteEntry,
43    RouteList,
44)
45from ..tags import Tag
46from ._common import (
47    TIMEOUT_FAST,
48    TIMEOUT_INTERACTIVE,
49    confirm_or_raise,
50    lean_schema_view,
51)
52
53if TYPE_CHECKING:
54    from music_assistant.mass import MusicAssistant
55
56
57LOGGER = logging.getLogger("music_assistant.providers.fastmcp_server.debug")
58
59_PAYLOAD_CAP_BYTES = 256 * 1024
60
61_TRACKED_PACKAGES = (
62    "music_assistant",
63    "music_assistant_models",
64    "fastmcp",
65    "aiohttp",
66    "mashumaro",
67)
68
69_RELOAD_POLL_SECONDS = 5.0
70_RELOAD_POLL_INTERVAL = 0.1
71
72
73def _safe_get(obj: Any, name: str, default: Any = None) -> Any:
74    """
75    Like getattr(obj, name, default) but also swallows property exceptions.
76
77    The inspect tools claim to work on broken/unavailable entities — a raising
78    @property on the inspected object must not crash the tool. The serializer
79    itself is defensive at the dataclass-field level; this helper covers the
80    one-shot tool-level accesses (state, manifest, current_item).
81    """
82    try:
83        return getattr(obj, name, default)
84    except Exception:
85        return default
86
87
88def _readonly(title: str) -> ToolAnnotations:
89    return ToolAnnotations(
90        title=title,
91        readOnlyHint=True,
92        destructiveHint=False,
93        idempotentHint=True,
94        openWorldHint=False,
95    )
96
97
98def _register_reload_tool(
99    sub: FastMCP, mass: MusicAssistant, *, require_confirmation: bool, reload_lock: asyncio.Lock
100) -> None:
101    @sub.tool(
102        tags={Tag.DEBUG_RELOAD},
103        annotations=ToolAnnotations(
104            title="Reload provider",
105            destructiveHint=True,
106            idempotentHint=False,
107        ),
108        timeout=TIMEOUT_INTERACTIVE,
109    )
110    async def reload_provider(instance_id: str, ctx: Context | None = None) -> ReloadResult:
111        """
112        Unload and reload a configured provider instance.
113
114        INTERRUPTS ACTIVE STREAMS on the affected provider. Confirmation is
115        required by default. See also: debug_inspect_provider to verify the
116        reload landed; debug_tail_log for the reload's own log lines.
117
118        :param instance_id: Provider instance identifier.
119        :param ctx: FastMCP context — populated automatically by the server.
120        """
121        await confirm_or_raise(
122            ctx,
123            (
124                f"Reload provider {instance_id!r}? "
125                "Active playback on this provider will be interrupted."
126            ),
127            enabled=require_confirmation,
128        )
129        try:
130            conf = await mass.config.get_provider_config(instance_id)
131        except Exception as exc:
132            raise ToolError(f"provider instance_id={instance_id!r} not configured") from exc
133
134        async with reload_lock:
135            LOGGER.info(
136                "MCP debug_reload_provider triggered: instance_id=%s",
137                instance_id,
138            )
139            t0 = time.monotonic()
140            load_error: Exception | None = None
141            try:
142                # Private but the only reload primitive — see spec 0005 "Known private-API carve-outs".
143                await mass._load_provider(conf)
144            except Exception as exc:
145                load_error = exc
146
147            if load_error is not None:
148                return ReloadResult(
149                    instance_id=instance_id,
150                    duration_ms=(time.monotonic() - t0) * 1000,
151                    new_available=False,
152                    last_error=str(load_error),
153                )
154
155            deadline = time.monotonic() + _RELOAD_POLL_SECONDS
156            prov = None
157            while time.monotonic() < deadline:
158                prov = mass.get_provider(instance_id)
159                if prov and getattr(prov, "available", False):
160                    break
161                await asyncio.sleep(_RELOAD_POLL_INTERVAL)
162            available = bool(prov and getattr(prov, "available", False))
163            last_error = getattr(prov, "last_error", None) if prov else "reload timed out"
164            return ReloadResult(
165                instance_id=instance_id,
166                duration_ms=(time.monotonic() - t0) * 1000,
167                new_available=available,
168                last_error=last_error,
169            )
170
171
172def _register_logs_tool(sub: FastMCP, mass: MusicAssistant) -> None:
173    tail = SafeLogTail(mass)
174
175    @sub.tool(
176        tags={Tag.DEBUG_LOGS},
177        annotations=_readonly("Tail MA log"),
178        timeout=TIMEOUT_FAST,
179    )  # type: ignore[untyped-decorator, unused-ignore]
180    async def tail_log(
181        lines: int = 200,
182        level: str | None = None,
183        component_regex: str | None = None,
184        search: str | None = None,
185        since_seconds: int | None = None,
186        before: str | None = None,
187        name: str = "musicassistant.log",
188    ) -> LogTailResult:
189        """
190        Return the last N matching log records with optional filters.
191
192        Multi-line records (tracebacks) are returned whole, and ``lines``
193        counts *matching records* — filters apply before the limit, so
194        ``lines=5, level="ERROR"`` is "the 5 most recent errors". When the
195        page is incomplete, ``has_more`` / ``response_truncated`` are set and
196        ``next_call_hint`` carries a ready-to-use follow-up call. To watch the
197        log "live", re-call periodically with ``since_seconds`` covering the
198        polling gap. Bearer tokens and common secret patterns are redacted.
199        See also: debug_log_stats for a cheap aggregate view before pulling
200        raw records; debug_recent_events for state transitions.
201
202        :param lines: Number of matching records to return (clamped to [1, 2000]).
203        :param level: Minimum severity, case-insensitive — e.g. ``"warning"``
204            returns WARNING, ERROR and CRITICAL records.
205        :param component_regex: Optional regex matched against the component name.
206        :param search: Optional case-insensitive regex matched against the full
207            record text, including traceback lines.
208        :param since_seconds: When set, only records within this many seconds
209            of "now" are returned.
210        :param before: Paging cursor — an ISO timestamp, or the exact
211            ``offset:<n>`` value from a previous result's ``next_call_hint``
212            (the offset form is lossless when many records share a timestamp).
213        :param name: Log file basename within ``$HOME/.musicassistant/``. Only the
214            canonical log and its rotated siblings (``.log.1`` … ``.log.5``) are
215            allowed.
216        """
217        # Offload the synchronous file scan (up to the 10 MB cap) to a worker
218        # thread so it never stalls MA's single event loop.
219        return await asyncio.to_thread(
220            tail.tail,
221            lines=lines,
222            level=level,
223            component_regex=component_regex,
224            search=search,
225            since_seconds=since_seconds,
226            before=before,
227            name=name,
228        )
229
230    @sub.tool(
231        tags={Tag.DEBUG_LOGS},
232        annotations=_readonly("Log statistics"),
233        timeout=TIMEOUT_FAST,
234    )  # type: ignore[untyped-decorator, unused-ignore]
235    async def log_stats(
236        since_seconds: int | None = None,
237        name: str = "musicassistant.log",
238    ) -> LogStatsResult:
239        """
240        Aggregate view of the log: record counts per level, top components, time range.
241
242        Use this before ``debug_tail_log`` to scope a problem cheaply — the
243        counts show whether (and where) errors exist without spending context
244        on raw lines. Scans at most 10 MB from the end of the file
245        (``truncated`` is set when the cap fires).
246
247        :param since_seconds: Restrict the window to the last N seconds.
248        :param name: Log file basename within ``$HOME/.musicassistant/``. Only the
249            canonical log and its rotated siblings (``.log.1`` … ``.log.5``) are
250            allowed.
251        """
252        return await asyncio.to_thread(tail.stats, since_seconds=since_seconds, name=name)
253
254
255def build_debug_server(
256    mass: MusicAssistant,
257    *,
258    require_confirmation: bool = True,
259    event_buffer: EventBuffer | None = None,
260    logs_enabled: bool = True,
261    reload_lock: asyncio.Lock | None = None,
262    lean_schema: bool = False,
263) -> FastMCP:
264    """
265    Build the ``debug`` sub-server.
266
267    :param mass: MusicAssistant instance.
268    :param require_confirmation: When True (default), ``debug_reload_provider``
269        elicits explicit confirmation from the MCP client before reloading.
270    :param event_buffer: A started ``EventBuffer`` instance. When ``None``
271        the events tools still mount but report ``current_size=0`` and
272        empty snapshots — useful for tests that only exercise other groups.
273    :param logs_enabled: Whether the ``DEBUG_LOGS`` capability is enabled. When
274        ``False``, ``debug_health_summary`` skips its log-error read and reports
275        ``DEBUG_LOGS`` as a disabled capability instead.
276    :param reload_lock: Lock serialising ``debug_reload_provider`` for this
277        runtime. Defaults to a fresh per-server lock so independent servers
278        (e.g. test instances) never serialise against one another.
279    :param lean_schema: When True, tools omit their ``outputSchema`` to shrink
280        the namespace's context footprint for hosts without tool-search.
281    """
282    sub = FastMCP(name="debug")
283    target = lean_schema_view(sub) if lean_schema else sub
284    _register_inspect_tools(target, mass)
285    _register_logs_tool(target, mass)
286    _register_events_tools(target, mass, event_buffer)
287    _register_providers_tools(target, mass)
288    _register_reload_tool(
289        target,
290        mass,
291        require_confirmation=require_confirmation,
292        reload_lock=reload_lock if reload_lock is not None else asyncio.Lock(),
293    )
294    _register_health_tool(target, mass, buffer=event_buffer, logs_enabled=logs_enabled)
295    return sub
296
297
298def _register_inspect_tools(sub: FastMCP, mass: MusicAssistant) -> None:
299    @sub.tool(
300        tags={Tag.DEBUG_INSPECT},
301        annotations=_readonly("Inspect raw player state"),
302        timeout=TIMEOUT_FAST,
303    )  # type: ignore[untyped-decorator, unused-ignore]
304    async def inspect_player(player_id: str) -> PlayerInspect:
305        """
306        Return the raw runtime state of a player, including state.* fields the brief omits.
307
308        Works for unavailable and disabled players — that is the point.
309        See also: debug_recent_events with id_filter=<player_id> for transitions,
310        debug_tail_log for the textual context.
311
312        :param player_id: Identifier of the player to inspect.
313        """
314        player = mass.players.get_player(player_id)
315        if player is None:
316            raise ToolError(f"player_id={player_id!r} not found")
317        state_obj = _safe_get(player, "state", None)
318        raw, raw_trunc = dump(
319            player,
320            max_total_bytes=_PAYLOAD_CAP_BYTES,
321            return_truncated=True,
322        )
323        if state_obj is not None:
324            state, state_trunc = dump(
325                state_obj,
326                max_total_bytes=_PAYLOAD_CAP_BYTES,
327                return_truncated=True,
328            )
329        else:
330            state, state_trunc = {}, False
331        # raw includes state; surface it again under .state for convenience.
332        if isinstance(raw, dict) and "state" in raw:
333            raw.pop("state", None)
334        return PlayerInspect(
335            player_id=player_id,
336            raw=raw,
337            state=state,
338            truncated=bool(raw_trunc or state_trunc),
339        )
340
341    @sub.tool(
342        tags={Tag.DEBUG_INSPECT},
343        annotations=_readonly("Inspect raw queue state"),
344        timeout=TIMEOUT_FAST,
345    )  # type: ignore[untyped-decorator, unused-ignore]
346    async def inspect_queue(queue_id: str) -> QueueInspect:
347        """
348        Return the raw runtime state of a PlayerQueue plus the current_item resolved.
349
350        See also: debug_inspect_player for the queue's owning player,
351        debug_recent_events with id_filter=<queue_id> for transitions.
352
353        :param queue_id: Identifier of the queue to inspect.
354        """
355        queue = mass.player_queues.get(queue_id)
356        if queue is None:
357            raise ToolError(f"queue_id={queue_id!r} not found")
358        raw, raw_trunc = dump(queue, max_total_bytes=_PAYLOAD_CAP_BYTES, return_truncated=True)
359        current = _safe_get(queue, "current_item", None)
360        current_payload, current_trunc = (
361            dump(current, max_total_bytes=_PAYLOAD_CAP_BYTES, return_truncated=True)
362            if current is not None
363            else (None, False)
364        )
365        return QueueInspect(
366            queue_id=queue_id,
367            raw=raw,
368            current_item=current_payload,
369            truncated=bool(raw_trunc or current_trunc),
370        )
371
372    @sub.tool(
373        tags={Tag.DEBUG_INSPECT},
374        annotations=_readonly("Inspect raw provider state"),
375        timeout=TIMEOUT_FAST,
376    )  # type: ignore[untyped-decorator, unused-ignore]
377    async def inspect_provider(instance_id: str) -> ProviderInspect:
378        """
379        Return the raw runtime state of a configured provider plus its manifest.
380
381        See also: debug_inspect_provider_config for masked configuration,
382        debug_list_webserver_routes for the routes this provider registered,
383        debug_reload_provider to restart it.
384
385        :param instance_id: Identifier of the provider instance to inspect.
386        """
387        prov = mass.get_provider(instance_id)
388        if prov is None:
389            raise ToolError(f"provider instance_id={instance_id!r} not configured")
390        manifest = _safe_get(prov, "manifest", None)
391        raw, raw_trunc = dump(prov, max_total_bytes=_PAYLOAD_CAP_BYTES, return_truncated=True)
392        manifest_payload, manifest_trunc = (
393            dump(manifest, max_total_bytes=_PAYLOAD_CAP_BYTES, return_truncated=True)
394            if manifest is not None
395            else ({}, False)
396        )
397        if isinstance(raw, dict):
398            raw.pop("manifest", None)
399        return ProviderInspect(
400            instance_id=instance_id,
401            raw=raw,
402            manifest=manifest_payload,
403            truncated=bool(raw_trunc or manifest_trunc),
404        )
405
406
407def _register_events_tools(
408    sub: FastMCP,
409    mass: MusicAssistant,  # noqa: ARG001 -- reserved for symmetry/future use
410    buffer: EventBuffer | None,
411) -> None:
412    @sub.tool(
413        tags={Tag.DEBUG_EVENTS},
414        annotations=_readonly("Read recent MA events"),
415        timeout=TIMEOUT_FAST,
416    )  # type: ignore[untyped-decorator, unused-ignore]
417    async def recent_events(
418        limit: int = 100,
419        event_types: list[str] | None = None,
420        id_filter: str | None = None,
421        since_seconds: int | None = None,
422    ) -> EventSnapshot:
423        """
424        Return the most recent events captured into the in-memory ring buffer.
425
426        See also: debug_tail_log for the textual context around an event timestamp.
427
428        :param limit: Maximum events to return (clamped to [1, 1000]).
429        :param event_types: Optional list of event-type strings to include.
430        :param id_filter: Optional ``object_id`` to filter by.
431        :param since_seconds: When set, only events within this many seconds of
432            "now" are returned.
433        """
434        if buffer is None:
435            return EventSnapshot(events=[], buffer_capacity=0, total_seen=0)
436        events = buffer.snapshot(
437            limit=limit,
438            event_types=event_types,
439            id_filter=id_filter,
440            since_seconds=since_seconds,
441        )
442        stats = buffer.stats()
443        return EventSnapshot(
444            events=events,
445            buffer_capacity=stats.capacity,
446            total_seen=stats.total_seen,
447        )
448
449    @sub.tool(
450        tags={Tag.DEBUG_EVENTS},
451        annotations=_readonly("Event buffer stats"),
452        timeout=TIMEOUT_FAST,
453    )  # type: ignore[untyped-decorator, unused-ignore]
454    async def event_buffer_stats() -> EventBufferStats:
455        """
456        Return introspection counters for the event ring buffer.
457
458        Use this to distinguish "no events match" from "events were dropped
459        before you asked". ``dropped`` is non-zero whenever the buffer overflowed.
460        """
461        if buffer is None:
462            return EventBufferStats(
463                capacity=0,
464                current_size=0,
465                total_seen=0,
466                dropped=0,
467                subscribed_since=None,
468                by_type={},
469            )
470        return buffer.stats()
471
472
473def _register_providers_tools(sub: FastMCP, mass: MusicAssistant) -> None:
474    @sub.tool(
475        tags={Tag.DEBUG_PROVIDERS},
476        annotations=_readonly("List configured providers"),
477        timeout=TIMEOUT_FAST,
478    )  # type: ignore[untyped-decorator, unused-ignore]
479    async def list_providers() -> ProviderList:
480        """
481        Roll-up of every configured provider.
482
483        See also: debug_inspect_provider for the full runtime dump,
484        debug_inspect_provider_config for the masked configuration,
485        debug_health_summary for triage.
486        """
487        summaries: list[ProviderSummary] = []
488        for prov in getattr(mass, "providers", []):
489            ptype = getattr(getattr(prov, "type", None), "value", None) or str(
490                getattr(prov, "type", "unknown")
491            )
492            summaries.append(
493                ProviderSummary(
494                    instance_id=getattr(prov, "instance_id", ""),
495                    domain=getattr(prov, "domain", ""),
496                    type=ptype,
497                    name=getattr(prov, "name", "") or getattr(prov, "domain", ""),
498                    available=bool(getattr(prov, "available", False)),
499                    last_error=getattr(prov, "last_error", None),
500                )
501            )
502        return ProviderList(providers=summaries)
503
504    @sub.tool(
505        tags={Tag.DEBUG_PROVIDERS},
506        annotations=_readonly("Inspect provider config (masked)"),
507        timeout=TIMEOUT_FAST,
508    )  # type: ignore[untyped-decorator, unused-ignore]
509    async def inspect_provider_config(instance_id: str) -> ProviderConfigDump:
510        """
511        Dump a provider's stored ConfigEntry values.
512
513        SECURE_STRING values are replaced by MA's SECURE_STRING_SUBSTITUTE
514        sentinel via ``__post_serialize__`` in ``music_assistant_models`` —
515        this tool carries no masking logic of its own.
516
517        :param instance_id: The provider instance identifier.
518        """
519        try:
520            config = await mass.config.get_provider_config(instance_id)
521        except Exception as exc:
522            raise ToolError(f"provider instance_id={instance_id!r} not configured") from exc
523        raw = config.to_dict()
524        values: list[ConfigValueDump] = []
525        truncated = False
526        running_bytes = 0
527        for key, entry in raw.get("values", {}).items():
528            value = entry.get("value")
529            etype = entry.get("type", "unknown")
530            payload_size = len(str(value)) + len(key) + len(etype) + 16
531            if running_bytes + payload_size > _PAYLOAD_CAP_BYTES:
532                truncated = True
533                break
534            running_bytes += payload_size
535            values.append(ConfigValueDump(key=str(key), type=str(etype), value=value))
536        return ProviderConfigDump(
537            instance_id=instance_id,
538            domain=raw.get("domain", ""),
539            values=values,
540            truncated=truncated,
541        )
542
543    @sub.tool(
544        tags={Tag.DEBUG_PROVIDERS},
545        annotations=_readonly("List webserver routes"),
546        timeout=TIMEOUT_FAST,
547    )  # type: ignore[untyped-decorator, unused-ignore]
548    async def list_webserver_routes() -> RouteList:
549        """
550        Enumerate the HTTP routes registered on MA's webserver.
551
552        Includes both dynamic (provider-registered) and static routes.
553        Reaches into ``webserver._server.app.router`` — single documented
554        private-API touch. See spec 0005.
555        """
556        routes: list[RouteEntry] = []
557        try:
558            inner_app = mass.webserver._server.app  # type: ignore[attr-defined]
559            for route in inner_app.router.routes():
560                method = str(getattr(route, "method", "*"))
561                resource = getattr(route, "resource", None)
562                path = str(getattr(resource, "canonical", "")) if resource else ""
563                routes.append(
564                    RouteEntry(
565                        method=method,
566                        path=path,
567                        registered_by=_attribute_route(path),
568                    )
569                )
570        except AttributeError as exc:
571            raise ToolError("webserver routes are unavailable in this MA build") from exc
572        return RouteList(routes=routes)
573
574    @sub.tool(
575        tags={Tag.DEBUG_PROVIDERS},
576        annotations=_readonly("List installed package versions"),
577        timeout=TIMEOUT_FAST,
578    )  # type: ignore[untyped-decorator, unused-ignore]
579    async def list_package_versions() -> PackageVersions:
580        """
581        Return installed versions of the key packages backing the MCP provider and MA.
582
583        Useful for upstream bug reports.
584        """
585        out: dict[str, str] = {}
586        for pkg in _TRACKED_PACKAGES:
587            try:
588                out[pkg] = importlib.metadata.version(pkg)
589            except importlib.metadata.PackageNotFoundError:
590                out[pkg] = "<not installed>"
591        return PackageVersions(packages=out)
592
593
594def _register_health_tool(
595    sub: FastMCP, mass: MusicAssistant, *, buffer: EventBuffer | None, logs_enabled: bool
596) -> None:
597    @sub.tool(
598        tags={Tag.DEBUG_PROVIDERS},
599        annotations=_readonly("Health summary roll-up"),
600        timeout=TIMEOUT_FAST,
601    )  # type: ignore[untyped-decorator, unused-ignore]
602    async def health_summary() -> HealthSummary:
603        """
604        Entry-point triage tool — one read returns a roll-up of provider state, queue counts, event rate, and log error count.
605
606        If a section flags errors, drill into: debug_inspect_provider for provider
607        errors, debug_inspect_queue for queue errors, debug_tail_log for the
608        recent ERROR log lines. Fields whose capability is disabled show as
609        ``None`` with the tag name listed in ``disabled_capabilities``.
610        """
611        providers = list(getattr(mass, "providers", []))
612        loaded = sum(1 for p in providers if getattr(p, "available", False))
613        disabled = sum(1 for p in providers if not getattr(p, "enabled", True))
614        error_details: list[ProviderSummary] = []
615        for p in providers:
616            if getattr(p, "last_error", None):
617                ptype = getattr(getattr(p, "type", None), "value", "unknown")
618                error_details.append(
619                    ProviderSummary(
620                        instance_id=getattr(p, "instance_id", ""),
621                        domain=getattr(p, "domain", ""),
622                        type=str(ptype),
623                        name=getattr(p, "name", "") or getattr(p, "domain", ""),
624                        available=bool(getattr(p, "available", False)),
625                        last_error=getattr(p, "last_error", None),
626                    )
627                )
628
629        try:
630            queues = list(mass.player_queues.all())
631        except AttributeError, TypeError:
632            queues = []
633        queues_active = sum(1 for q in queues if getattr(q, "state", None) == "playing")
634        queues_errors = sum(
635            1
636            for q in queues
637            if getattr(q, "state", None) == "error" or not getattr(q, "available", True)
638        )
639
640        disabled_capabilities: list[str] = []
641        events_per_min: dict[str, float] | None = None
642        if buffer is None:
643            disabled_capabilities.append("DEBUG_EVENTS")
644        else:
645            stats = buffer.stats()
646            if stats.subscribed_since is None:
647                disabled_capabilities.append("DEBUG_EVENTS")
648            else:
649                from datetime import datetime  # noqa: PLC0415
650
651                from music_assistant.helpers.datetime import now as ma_now  # noqa: PLC0415
652
653                subscribed_at = datetime.fromisoformat(stats.subscribed_since)
654                elapsed_min = max(
655                    1.0 / 60,
656                    (ma_now() - subscribed_at).total_seconds() / 60.0,
657                )
658                events_per_min = {
659                    et: round(count / elapsed_min, 2) for et, count in stats.by_type.items()
660                }
661
662        log_errors: int | None = None
663        if not logs_enabled:
664            # DEBUG_LOGS is off — do not read the log file (mirrors the events
665            # gate above; reading here would bypass the disabled permission).
666            disabled_capabilities.append("DEBUG_LOGS")
667        else:
668            try:
669                # Offload the synchronous scan off MA's event loop (see tail_log).
670                log_errors = await asyncio.to_thread(SafeLogTail(mass).count_errors_last_5min)
671            except Exception:
672                disabled_capabilities.append("DEBUG_LOGS")
673
674        return HealthSummary(
675            providers_loaded=loaded,
676            providers_disabled=disabled,
677            providers_error=len(error_details),
678            providers_error_details=error_details,
679            queues_total=len(queues),
680            queues_with_active_playback=queues_active,
681            queues_with_errors=queues_errors,
682            events_per_min_by_type=events_per_min,
683            log_errors_last_5min=log_errors,
684            disabled_capabilities=disabled_capabilities,
685        )
686
687
688def _attribute_route(path: str) -> str | None:
689    """Best-effort: map a route path back to who registered it by path prefix."""
690    if path.startswith("/mcp/"):
691        return "fastmcp_server"
692    if path.startswith("/.well-known/"):
693        return "fastmcp_server (well-known)"
694    if path.startswith("/api/"):
695        return "music_assistant (api)"
696    return None
697