fc3i: /api/system/activity blueprint — queues, workers, runs, failures

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

backend/app/api/__init__.py

@@ -33,6 +33,7 @@ def all_blueprints() -> list[Blueprint]:
     from .showcase import showcase_bp
     from .sources import sources_bp
     from .suggestions import suggestions_bp
+    from .system_activity import system_activity_bp
     from .tags import tags_bp
     return [
         api_bp,
@@ -44,6 +45,7 @@ def all_blueprints() -> list[Blueprint]:
         artists_bp,
         showcase_bp,
         settings_bp,
+        system_activity_bp,
         import_admin_bp,
         migrate_bp,
         suggestions_bp,

backend/app/api/system_activity.py

@@ -0,0 +1,209 @@
+"""FC-3i: system activity dashboard endpoints.
+
+Read-only. Combines Redis-broker queue depths (LLEN per queue),
+Celery worker introspection (celery inspect), and the task_run DB
+history into the surfaces the SystemActivityTab UI consumes.
+
+All filesystem/sync-client work goes through asyncio.to_thread per
+ASYNC230/240 (mirrors backend.app.api.extension's pattern).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from datetime import UTC, datetime, timedelta
+
+from quart import Blueprint, jsonify, request
+from sqlalchemy import desc, func, select
+
+from ..config import get_config
+from ..extensions import get_session
+from ..models import TaskRun
+
+system_activity_bp = Blueprint(
+    "system_activity", __name__, url_prefix="/api/system/activity",
+)
+
+# Canonical queue order — must match celery_app.task_routes. UI renders
+# in this order; queues with no LLEN response show as null rather than
+# absent.
+_QUEUE_NAMES = (
+    "default", "import", "thumbnail", "ml",
+    "download", "scan", "maintenance",
+)
+
+# Cache module-level so all requests share the cache between polls.
+# Tests can reset via direct dict mutation if needed.
+_QUEUE_CACHE: dict = {"ts": 0.0, "data": None}
+_WORKER_CACHE: dict = {"ts": 0.0, "data": None}
+_QUEUE_CACHE_TTL = 2.0
+_WORKER_CACHE_TTL = 5.0
+
+
+def _read_queues_sync() -> dict:
+    """Reads each queue's LLEN from the broker. Sync — caller wraps in
+    asyncio.to_thread. Per-queue try/except returns None on failure so
+    one bad queue doesn't break the whole response."""
+    import redis  # local import; only this endpoint needs it
+
+    cfg = get_config()
+    client = redis.Redis.from_url(cfg.celery_broker_url)
+    out: dict = {}
+    for name in _QUEUE_NAMES:
+        try:
+            out[name] = int(client.llen(name))
+        except Exception:  # noqa: BLE001 — broker hiccup shouldn't break UI
+            out[name] = None
+    return {
+        "queues": out,
+        "fetched_at": datetime.now(UTC).isoformat(),
+    }
+
+
+def _read_workers_sync() -> dict:
+    """celery inspect active_queues + active. Returns per-worker info."""
+    from ..celery_app import celery as celery_app
+
+    insp = celery_app.control.inspect(timeout=2.0)
+    active_queues = insp.active_queues() or {}
+    active_tasks = insp.active() or {}
+
+    workers: dict = {}
+    for hostname, queues in active_queues.items():
+        workers[hostname] = {
+            "queues": sorted({q["name"] for q in queues}),
+            "active_count": len(active_tasks.get(hostname, [])),
+        }
+    return {
+        "workers": workers,
+        "fetched_at": datetime.now(UTC).isoformat(),
+    }
+
+
+@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"])
+
+
+@system_activity_bp.route("/workers", methods=["GET"])
+async def get_workers():
+    """Live celery inspect. Cached 5s.
+
+    Response: {workers: {hostname: {queues, active_count}}, fetched_at}
+    """
+    now = time.time()
+    if _WORKER_CACHE["data"] is None or (now - _WORKER_CACHE["ts"]) > _WORKER_CACHE_TTL:
+        _WORKER_CACHE["data"] = await asyncio.to_thread(_read_workers_sync)
+        _WORKER_CACHE["ts"] = now
+    return jsonify(_WORKER_CACHE["data"])
+
+
+@system_activity_bp.route("/runs", methods=["GET"])
+async def list_runs():
+    """Paginated task_run history. Query params:
+      queue=<name>     filter to one queue
+      status=<status>  filter to one status (running/ok/error/timeout/retry)
+      limit=<int>      default 50, max 200
+      before_id=<int>  cursor for keyset pagination
+
+    Response: {runs: [...], next_cursor: id|null}
+    """
+    try:
+        limit = min(int(request.args.get("limit", "50")), 200)
+    except ValueError:
+        return jsonify({"error": "invalid_limit"}), 400
+    if limit < 1:
+        return jsonify({"error": "invalid_limit"}), 400
+
+    queue = request.args.get("queue")
+    status = request.args.get("status")
+    before_id_raw = request.args.get("before_id")
+    before_id = int(before_id_raw) if before_id_raw else None
+
+    async with get_session() as session:
+        stmt = select(TaskRun).order_by(desc(TaskRun.id))
+        if queue:
+            stmt = stmt.where(TaskRun.queue == queue)
+        if status:
+            stmt = stmt.where(TaskRun.status == status)
+        if before_id is not None:
+            stmt = stmt.where(TaskRun.id < before_id)
+        stmt = stmt.limit(limit + 1)
+        rows = (await session.execute(stmt)).scalars().all()
+
+    has_more = len(rows) > limit
+    rows = rows[:limit]
+    return jsonify({
+        "runs": [_row_to_dict(r) for r in rows],
+        "next_cursor": rows[-1].id if has_more and rows else None,
+    })
+
+
+@system_activity_bp.route("/failures", methods=["GET"])
+async def list_failures():
+    """Recent failures across all lanes (24h window).
+
+    Response: {recent: [...], count_by_type: {ErrorClass: n}, since}
+    """
+    try:
+        limit = min(int(request.args.get("limit", "50")), 200)
+    except ValueError:
+        return jsonify({"error": "invalid_limit"}), 400
+
+    since = datetime.now(UTC) - timedelta(hours=24)
+
+    async with get_session() as session:
+        recent_stmt = (
+            select(TaskRun)
+            .where(TaskRun.status.in_(["error", "timeout"]))
+            .where(TaskRun.finished_at >= since)
+            .order_by(desc(TaskRun.finished_at))
+            .limit(limit)
+        )
+        recent = (await session.execute(recent_stmt)).scalars().all()
+
+        count_stmt = (
+            select(TaskRun.error_type, func.count(TaskRun.id))
+            .where(TaskRun.status.in_(["error", "timeout"]))
+            .where(TaskRun.finished_at >= since)
+            .group_by(TaskRun.error_type)
+            .order_by(desc(func.count(TaskRun.id)))
+        )
+        counts = (await session.execute(count_stmt)).all()
+
+    return jsonify({
+        "recent": [_row_to_dict(r) for r in recent],
+        "count_by_type": {
+            (row[0] or "Unknown"): row[1]
+            for row in counts
+        },
+        "since": since.isoformat(),
+    })
+
+
+def _row_to_dict(r: TaskRun) -> dict:
+    return {
+        "id": r.id,
+        "queue": r.queue,
+        "task_name": r.task_name,
+        "target_id": r.target_id,
+        "celery_task_id": r.celery_task_id,
+        "started_at": r.started_at.isoformat() if r.started_at else None,
+        "finished_at": r.finished_at.isoformat() if r.finished_at else None,
+        "duration_ms": r.duration_ms,
+        "status": r.status,
+        "error_type": r.error_type,
+        "error_message": r.error_message,
+        "retry_count": r.retry_count,
+        "worker_hostname": r.worker_hostname,
+        "args_summary": r.args_summary,
+    }