Show actual collector search activity in request pipeline

backend/app/services/collector_search.py

@@ -0,0 +1,61 @@
+"""Read title-specific search activity without starting a search or changing monitoring."""
+
+from typing import Any
+
+from ..clients.base import ApiClient
+from ..models import RequestType
+
+
+def _ids(values: Any) -> set[int]:
+    if not isinstance(values, list):
+        return set()
+    return {value for value in values if type(value) is int and value > 0}
+
+
+def search_status(commands: Any, request_type: RequestType, item_id: int, episodes: Any = None) -> str:
+    """Only a matching queued/started search is evidence of current activity.
+
+    Completed commands, RSS syncs and library-wide jobs do not establish that this
+    title is being searched. Episode searches are matched using Sonarr episode IDs.
+    """
+    if not isinstance(commands, list):
+        return "unavailable"
+    episode_ids = _ids([
+        episode.get("id") for episode in (episodes if isinstance(episodes, list) else [])
+        if isinstance(episode, dict) and episode.get("seriesId", item_id) == item_id
+    ])
+    queued = False
+    for command in commands:
+        if not isinstance(command, dict):
+            continue
+        body = command.get("body")
+        if not isinstance(body, dict):
+            continue
+        name = str(command.get("name") or body.get("name") or "").lower()
+        if request_type == RequestType.movie:
+            matches = name == "moviessearch" and item_id in _ids(body.get("movieIds"))
+        else:
+            matches = (
+                name in {"seriessearch", "seasonsearch"} and body.get("seriesId") == item_id
+            ) or (
+                name == "episodesearch" and bool(episode_ids & _ids(body.get("episodeIds")))
+            )
+        if not matches or command.get("ended"):
+            continue
+        status = str(command.get("status", "")).lower()
+        if status in {"started", "1"}:
+            return "searching"
+        if status in {"queued", "0"}:
+            queued = True
+    return "queued" if queued else "idle"
+
+
+async def read_search_status(
+    client: ApiClient, request_type: RequestType, item_id: int, episodes: Any = None,
+) -> str:
+    try:
+        commands = await client.get("/api/v3/command", timeout_seconds=3.0)
+    except Exception:
+        # Search telemetry must not turn a healthy library record into an error.
+        return "unavailable"
+    return search_status(commands, request_type, item_id, episodes)