feat(I3): always-on pipeline status indicator in the top nav

New /api/system/activity/summary aggregates scheduler health + per-queue
pending depths + running count + 24h failure count in one cached call (safe
to poll app-wide). PipelineStatusChip lives in the TopNav on every page: a
compact running/queued/failing chip with a scheduler-health dot that expands
to a popover (scheduler, busy queues, counts, link to downloads). Polls the
summary every 8s, paused when backgrounded. Reuses the queue-cache read via
_queues_cached(). + API test for the summary shape.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/api/system_activity.py

@@ -20,6 +20,7 @@ from sqlalchemy import desc, func, select
 from ..config import get_config
 from ..extensions import get_session
 from ..models import TaskRun
+from ..services.scheduler_service import scheduler_status
 
 system_activity_bp = Blueprint(
     "system_activity", __name__, url_prefix="/api/system/activity",
@@ -81,17 +82,22 @@ def _read_workers_sync() -> dict:
     }
 
 
+async def _queues_cached() -> dict:
+    """Per-queue Redis LLEN, cached 2s. Shared by /queues and /summary."""
+    now = time.time()
+    if _QUEUE_CACHE["data"] is None or (now - _QUEUE_CACHE["ts"]) > _QUEUE_CACHE_TTL:
+        _QUEUE_CACHE["data"] = await asyncio.to_thread(_read_queues_sync)
+        _QUEUE_CACHE["ts"] = now
+    return _QUEUE_CACHE["data"]
+
+
 @system_activity_bp.route("/queues", methods=["GET"])
 async def get_queues():
     """Per-queue Redis LLEN. Cached 2s.
 
     Response: {queues: {name: depth_or_null}, fetched_at: iso8601}
     """
-    now = time.time()
-    if _QUEUE_CACHE["data"] is None or (now - _QUEUE_CACHE["ts"]) > _QUEUE_CACHE_TTL:
-        _QUEUE_CACHE["data"] = await asyncio.to_thread(_read_queues_sync)
-        _QUEUE_CACHE["ts"] = now
-    return jsonify(_QUEUE_CACHE["data"])
+    return jsonify(await _queues_cached())
 
 
 @system_activity_bp.route("/workers", methods=["GET"])
@@ -107,6 +113,35 @@ async def get_workers():
     return jsonify(_WORKER_CACHE["data"])
 
 
+@system_activity_bp.route("/summary", methods=["GET"])
+async def get_summary():
+    """One-call rollup for the always-on TopNav pipeline indicator:
+    scheduler health, per-queue pending depths, currently-running count, and
+    recent (24h) failure count. Cheap — cached queue LLENs + two TaskRun
+    counts — so it's safe to poll app-wide."""
+    queues_data = await _queues_cached()
+    depths = queues_data.get("queues", {})
+    queued_total = sum(v for v in depths.values() if isinstance(v, int))
+    since = datetime.now(UTC) - timedelta(hours=24)
+    async with get_session() as session:
+        scheduler = await scheduler_status(session)
+        running = (await session.execute(
+            select(func.count(TaskRun.id)).where(TaskRun.status == "running")
+        )).scalar_one()
+        failing = (await session.execute(
+            select(func.count(TaskRun.id))
+            .where(TaskRun.status.in_(["error", "timeout"]))
+            .where(TaskRun.finished_at >= since)
+        )).scalar_one()
+    return jsonify({
+        "scheduler": scheduler,
+        "queues": depths,
+        "queued_total": queued_total,
+        "running": int(running),
+        "failing": int(failing),
+    })
+
+
 @system_activity_bp.route("/runs", methods=["GET"])
 async def list_runs():
     """Paginated task_run history. Query params: