feat(ingestion,frontend): add live scholar progress counter to dashboard

Emit scholar_progress SSE events during two-pass ingestion so the
Activity Monitor shows "X / N visited · Y finished · Z new publications"
in real time instead of static post-run counts.

- scholar_processing: add on_progress callback to _run_first_pass and
  _run_depth_pass; build _emit closure in run_scholar_iteration that
  publishes visited/finished/total via run_events; batch-emit scholars
  finished in first pass before depth pass; skip emission on abort/cancel
- run_status store: add scholarProgress state, reset on stream open and
  reset(); subscribe to scholar_progress SSE events with safe parsing
- DashboardPage: show live counter during run, fall back to static text
  when run is not active or no progress received yet

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/services/ingestion/scholar_processing.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import random
+from collections.abc import Callable, Coroutine
 from datetime import UTC, datetime
 from typing import Any
 
@@ -274,6 +275,7 @@ async def _run_first_pass(
     request_delay_seconds: int,
     queue_delay_seconds: int,
     progress: RunProgress,
+    on_progress: Callable[[int, int], Coroutine[Any, Any, None]] | None = None,
 ) -> dict[int, int]:
     first_pass_cstarts: dict[int, int] = {}
     for index, scholar in enumerate(scholars):
@@ -310,6 +312,8 @@ async def _run_first_pass(
                 scholars_remaining=len(scholars) - index - 1,
             )
             return first_pass_cstarts
+        if on_progress is not None:
+            await on_progress(1, 0)
         resume_cstart = outcome.result_entry.get("continuation_cstart")
         if resume_cstart is not None and int(resume_cstart) > start_cstart:
             first_pass_cstarts[int(scholar.id)] = int(resume_cstart)
@@ -330,6 +334,7 @@ async def _run_depth_pass(
     auto_queue_continuations: bool,
     queue_delay_seconds: int,
     progress: RunProgress,
+    on_progress: Callable[[int, int], Coroutine[Any, Any, None]] | None = None,
 ) -> None:
     for index, scholar in enumerate(scholars):
         resume_cstart = first_pass_cstarts.get(int(scholar.id))
@@ -367,6 +372,8 @@ async def _run_depth_pass(
                 scholars_remaining=len(scholars) - index - 1,
             )
             break
+        if on_progress is not None:
+            await on_progress(0, 1)
 
 
 async def run_scholar_iteration(
@@ -387,6 +394,8 @@ async def run_scholar_iteration(
     auto_queue_continuations: bool,
     queue_delay_seconds: int,
 ) -> RunProgress:
+    from app.services.runs.events import run_events
+
     progress = RunProgress()
     scholar_kwargs: dict[str, Any] = {
         "request_delay_seconds": request_delay_seconds,
@@ -396,6 +405,21 @@ async def run_scholar_iteration(
         "rate_limit_backoff_seconds": rate_limit_backoff_seconds,
         "page_size": page_size,
     }
+
+    visited = 0
+    finished = 0
+    total = len(scholars)
+
+    async def _emit(v: int = 0, f: int = 0) -> None:
+        nonlocal visited, finished
+        visited += v
+        finished += f
+        await run_events.publish(
+            run.id,
+            "scholar_progress",
+            {"visited": visited, "finished": finished, "total": total},
+        )
+
     first_pass_cstarts = await _run_first_pass(
         db_session,
         scholars=scholars,
@@ -407,8 +431,12 @@ async def run_scholar_iteration(
         request_delay_seconds=request_delay_seconds,
         queue_delay_seconds=queue_delay_seconds,
         progress=progress,
+        on_progress=_emit,
     )
     remaining_max = max(max_pages_per_scholar - 1, 0)
+    scholars_finished_in_first_pass = len(scholars) - len(first_pass_cstarts)
+    if scholars_finished_in_first_pass > 0:
+        await _emit(f=scholars_finished_in_first_pass)
     if remaining_max <= 0:
         return progress
     await _run_depth_pass(
@@ -424,5 +452,6 @@ async def run_scholar_iteration(
         auto_queue_continuations=auto_queue_continuations,
         queue_delay_seconds=queue_delay_seconds,
         progress=progress,
+        on_progress=_emit,
     )
     return progress

frontend/src/pages/DashboardPage.vue

@@ -486,9 +486,16 @@ watch(
                       </RouterLink>
                     </div>
                     <p class="mt-2 text-sm text-secondary">
-                      Started {{ formatDate(displayedLatestRun.start_dt) }}. Processed
+                      Started {{ formatDate(displayedLatestRun.start_dt) }}.
-                      {{ displayedLatestRun.scholar_count }} scholars and discovered
+                      <template v-if="runStatus.isLikelyRunning && runStatus.scholarProgress">
+                        {{ runStatus.scholarProgress.visited }} / {{ runStatus.scholarProgress.total }} visited
+                        · {{ runStatus.scholarProgress.finished }} finished
+                        · {{ displayedLatestRun.new_publication_count }} new publications.
+                      </template>
+                      <template v-else>
+                        Processed {{ displayedLatestRun.scholar_count }} scholars and discovered
                         {{ displayedLatestRun.new_publication_count }} new publications.
+                      </template>
                     </p>
                   </div>
                   <AppEmptyState

frontend/src/stores/run_status.ts

@@ -168,6 +168,7 @@ export const useRunStatusStore = defineStore("runStatus", {
     lastErrorRequestId: null as string | null,
     lastSyncAt: null as number | null,
     livePublications: [] as Array<PublicationItem>,
+    scholarProgress: null as { visited: number; finished: number; total: number } | null,
   }),
   getters: {
     isRunActive(state): boolean {
@@ -257,6 +258,7 @@ export const useRunStatusStore = defineStore("runStatus", {
         }
         activeStreamRunId = targetRunId;
         this.livePublications = [];
+        this.scholarProgress = null;
         eventSource = new EventSource(`/api/v1/runs/${targetRunId}/stream`);
         eventSource.addEventListener("publication_discovered", (e) => {
           try {
@@ -312,6 +314,19 @@ export const useRunStatusStore = defineStore("runStatus", {
             console.error("Failed to parse SSE event", err);
           }
         });
+        eventSource.addEventListener("scholar_progress", (e) => {
+          try {
+            const data = JSON.parse(e.data);
+            const visited = typeof data.visited === "number" ? Math.max(0, Math.trunc(data.visited)) : null;
+            const finished = typeof data.finished === "number" ? Math.max(0, Math.trunc(data.finished)) : null;
+            const total = typeof data.total === "number" ? Math.max(1, Math.trunc(data.total)) : null;
+            if (visited !== null && finished !== null && total !== null) {
+              this.scholarProgress = { visited, finished, total };
+            }
+          } catch (err) {
+            console.error("Failed to parse SSE event", err);
+          }
+        });
         eventSource.onerror = () => {
           // Reconnecting is handled automatically by EventSource,
           // but if it's permanently closed, we could do something here.
@@ -491,6 +506,7 @@ export const useRunStatusStore = defineStore("runStatus", {
       this.lastSyncAt = null;
       this.safetyState = createDefaultSafetyState();
       this.livePublications = [];
+      this.scholarProgress = null;
     },
   },
 });