From 00e2608ba1eb10c7ab313e8724f34f7b299dcc77 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Fri, 29 May 2026 13:24:11 -0400
Subject: [PATCH] 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            |  45 +++++-
 .../src/components/PipelineStatusChip.vue     | 146 ++++++++++++++++++
 frontend/src/components/TopNav.vue            |   2 +
 frontend/src/stores/systemActivity.js         |  16 +-
 tests/test_api_system_activity.py             |  19 +++
 5 files changed, 221 insertions(+), 7 deletions(-)
 create mode 100644 frontend/src/components/PipelineStatusChip.vue

diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py
index 480b8c9..0d0600c 100644
--- a/backend/app/api/system_activity.py
+++ b/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:
diff --git a/frontend/src/components/PipelineStatusChip.vue b/frontend/src/components/PipelineStatusChip.vue
new file mode 100644
index 0000000..0b026c0
--- /dev/null
+++ b/frontend/src/components/PipelineStatusChip.vue
@@ -0,0 +1,146 @@
+<template>
+  <v-menu location="bottom start" :close-on-content-click="false">
+    <template #activator="{ props }">
+      <button
+        v-bind="props" type="button"
+        class="fc-pulse" :class="`fc-pulse--${schedHealth}`"
+        :aria-label="`pipeline: ${running} running, ${queued} queued, ${failing} failing`"
+      >
+        <span class="fc-pulse__dot" />
+        <span v-if="running" class="fc-pulse__stat">
+          <v-icon size="13">mdi-progress-download</v-icon>{{ running }}
+        </span>
+        <span v-if="queued" class="fc-pulse__stat">
+          <v-icon size="13">mdi-tray-full</v-icon>{{ queued }}
+        </span>
+        <span v-if="failing" class="fc-pulse__stat fc-pulse__stat--err">
+          <v-icon size="13">mdi-alert-circle</v-icon>{{ failing }}
+        </span>
+        <span v-if="!running && !queued && !failing" class="fc-pulse__idle">idle</span>
+      </button>
+    </template>
+
+    <v-card min-width="300" class="fc-pulse__panel pa-3">
+      <div class="fc-pulse__row">
+        <span class="fc-pulse__rowdot" :class="`fc-pulse--${schedHealth}`" />
+        <strong>Scheduler</strong>
+        <v-spacer />
+        <span class="fc-pulse__muted">{{ schedLabel }}</span>
+      </div>
+
+      <div class="fc-pulse__sec">
+        <div class="fc-pulse__sechead">Queues</div>
+        <div v-if="busyQueues.length === 0" class="fc-pulse__muted">All idle</div>
+        <div v-for="q in busyQueues" :key="q.name" class="fc-pulse__qrow">
+          <span>{{ q.name }}</span><v-spacer /><strong>{{ q.depth }}</strong>
+        </div>
+      </div>
+
+      <div class="fc-pulse__sec fc-pulse__grid">
+        <div><span class="fc-pulse__muted">Running</span><div class="fc-pulse__big">{{ running }}</div></div>
+        <div><span class="fc-pulse__muted">Queued</span><div class="fc-pulse__big">{{ queued }}</div></div>
+        <div>
+          <span class="fc-pulse__muted">Failures 24h</span>
+          <div class="fc-pulse__big" :class="{ 'fc-pulse__big--err': failing }">{{ failing }}</div>
+        </div>
+      </div>
+
+      <RouterLink
+        class="fc-pulse__link"
+        :to="{ path: '/subscriptions', query: { tab: 'downloads' } }"
+      >Open downloads →</RouterLink>
+    </v-card>
+  </v-menu>
+</template>
+
+<script setup>
+import { computed, onMounted, onUnmounted } from 'vue'
+
+import { useSystemActivityStore } from '../stores/systemActivity.js'
+import { formatRelative } from '../utils/date.js'
+
+const store = useSystemActivityStore()
+
+// Beat ticks every 60s; flag the scheduler stale past ~3 min.
+const STALE_MS = 180_000
+const summary = computed(() => store.summary)
+const running = computed(() => summary.value?.running ?? 0)
+const queued = computed(() => summary.value?.queued_total ?? 0)
+const failing = computed(() => summary.value?.failing ?? 0)
+
+const schedHealth = computed(() => {
+  const t = summary.value?.scheduler?.last_tick_at
+  if (!t) return 'unknown'
+  return (Date.now() - new Date(t).getTime()) <= STALE_MS ? 'ok' : 'stale'
+})
+const schedLabel = computed(() => {
+  const s = summary.value?.scheduler
+  if (!s) return '—'
+  const ran = formatRelative(s.last_tick_at, { nullText: 'never' })
+  if (s.due_now > 0) return `ran ${ran} · ${s.due_now} due`
+  return `ran ${ran}`
+})
+const busyQueues = computed(() => {
+  const q = summary.value?.queues || {}
+  return Object.entries(q)
+    .filter(([, depth]) => typeof depth === 'number' && depth > 0)
+    .map(([name, depth]) => ({ name, depth }))
+})
+
+const POLL_MS = 8000
+let timer = null
+onMounted(() => {
+  store.loadSummary()
+  timer = setInterval(() => {
+    if (!document.hidden) store.loadSummary()
+  }, POLL_MS)
+})
+onUnmounted(() => { if (timer) clearInterval(timer) })
+</script>
+
+<style scoped>
+.fc-pulse {
+  display: inline-flex; align-items: center; gap: 6px;
+  padding: 3px 8px; border-radius: 999px;
+  background: rgb(var(--v-theme-on-surface) / 0.08);
+  color: rgb(var(--v-theme-on-surface));
+  font-size: 0.78rem; font-variant-numeric: tabular-nums;
+  cursor: pointer; border: 0;
+}
+.fc-pulse:hover { background: rgb(var(--v-theme-on-surface) / 0.16); }
+.fc-pulse__dot {
+  width: 8px; height: 8px; border-radius: 50%; flex: 0 0 auto;
+  background: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-pulse--ok .fc-pulse__dot,
+.fc-pulse--ok.fc-pulse__rowdot { background: rgb(var(--v-theme-success)); }
+.fc-pulse--stale .fc-pulse__dot,
+.fc-pulse--stale.fc-pulse__rowdot { background: rgb(var(--v-theme-error)); }
+.fc-pulse__stat { display: inline-flex; align-items: center; gap: 2px; }
+.fc-pulse__stat--err { color: rgb(var(--v-theme-error)); }
+.fc-pulse__idle {
+  text-transform: uppercase; letter-spacing: 0.06em;
+  color: rgb(var(--v-theme-on-surface-variant)); font-size: 0.68rem;
+}
+
+.fc-pulse__panel { background: rgb(var(--v-theme-surface)); }
+.fc-pulse__row { display: flex; align-items: center; gap: 8px; }
+.fc-pulse__rowdot {
+  width: 9px; height: 9px; border-radius: 50%;
+  background: rgb(var(--v-theme-on-surface-variant));
+}
+.fc-pulse__muted { color: rgb(var(--v-theme-on-surface-variant)); font-size: 0.8rem; }
+.fc-pulse__sec { margin-top: 12px; }
+.fc-pulse__sechead {
+  font-size: 0.72rem; text-transform: uppercase; letter-spacing: 0.05em;
+  color: rgb(var(--v-theme-on-surface-variant)); margin-bottom: 4px;
+}
+.fc-pulse__qrow { display: flex; align-items: center; font-size: 0.85rem; padding: 2px 0; }
+.fc-pulse__grid { display: flex; gap: 16px; }
+.fc-pulse__big { font-size: 1.3rem; font-weight: 700; font-variant-numeric: tabular-nums; }
+.fc-pulse__big--err { color: rgb(var(--v-theme-error)); }
+.fc-pulse__link {
+  display: inline-block; margin-top: 14px;
+  color: rgb(var(--v-theme-accent)); text-decoration: none; font-size: 0.85rem;
+}
+</style>
diff --git a/frontend/src/components/TopNav.vue b/frontend/src/components/TopNav.vue
index b2d7939..07b5983 100644
--- a/frontend/src/components/TopNav.vue
+++ b/frontend/src/components/TopNav.vue
@@ -8,6 +8,7 @@
       <span class="fc-health" :title="health.label">
         <v-icon size="x-small" :color="health.color">{{ health.icon }}</v-icon>
       </span>
+      <PipelineStatusChip />
     </div>
 
     <nav class="fc-links">
@@ -29,6 +30,7 @@
 import { computed, onMounted } from 'vue'
 import router, { FRONT_DOOR } from '../router.js'
 import { useSystemStore } from '../stores/system.js'
+import PipelineStatusChip from './PipelineStatusChip.vue'
 
 const system = useSystemStore()
 onMounted(() => system.refreshHealth())
diff --git a/frontend/src/stores/systemActivity.js b/frontend/src/stores/systemActivity.js
index e152ade..7d8252f 100644
--- a/frontend/src/stores/systemActivity.js
+++ b/frontend/src/stores/systemActivity.js
@@ -93,11 +93,23 @@ export const useSystemActivityStore = defineStore('systemActivity', () => {
     runsFilter.value = { ...runsFilter.value, ...filter }
   }
 
+  // One-call rollup for the always-on TopNav pipeline indicator:
+  // { scheduler, queues, queued_total, running, failing }.
+  const summary = ref(null)
+  async function loadSummary() {
+    try {
+      summary.value = await api.get('/api/system/activity/summary')
+    } catch (e) {
+      lastError.value = e.message
+    }
+    return summary.value
+  }
+
   return {
-    queues, workers, recentMinute, failures,
+    queues, workers, recentMinute, failures, summary,
     runs, runsCursor, runsHasMore, runsFilter,
     loading, lastError,
     loadQueues, loadWorkers, loadRecentMinute,
-    loadRuns, loadFailures, setFilter,
+    loadRuns, loadFailures, loadSummary, setFilter,
   }
 })
diff --git a/tests/test_api_system_activity.py b/tests/test_api_system_activity.py
index 0a493ab..f5832ef 100644
--- a/tests/test_api_system_activity.py
+++ b/tests/test_api_system_activity.py
@@ -44,6 +44,25 @@ async def test_queues_returns_all_known_queues(client, monkeypatch):
     assert set(body["queues"].keys()) == set(activity_module._QUEUE_NAMES)
 
 
+@pytest.mark.asyncio
+async def test_summary_returns_rollup_shape(client, monkeypatch):
+    def _fake():
+        return {
+            "queues": {**dict.fromkeys(activity_module._QUEUE_NAMES, 0), "ml": 3},
+            "fetched_at": datetime.now(UTC).isoformat(),
+        }
+    monkeypatch.setattr(activity_module, "_read_queues_sync", _fake)
+
+    resp = await client.get("/api/system/activity/summary")
+    assert resp.status_code == 200
+    body = await resp.get_json()
+    assert set(body.keys()) == {"scheduler", "queues", "queued_total", "running", "failing"}
+    assert body["queued_total"] == 3  # only ml has a non-zero depth
+    assert isinstance(body["running"], int)
+    assert isinstance(body["failing"], int)
+    assert set(body["scheduler"]) == {"last_tick_at", "next_due_at", "due_now", "auto_sources"}
+
+
 @pytest.mark.asyncio
 async def test_queues_cached_within_ttl(client, monkeypatch):
     call_count = {"n": 0}