From 4620978dcee974b873ebca4801e32356c8186623 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Tue, 3 Mar 2026 17:36:43 +0100 Subject: [PATCH 01/15] feat(ingestion,frontend): add live scholar progress counter to dashboard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/services/ingestion/scholar_processing.py | 29 ++++++++++++++++++++ frontend/src/pages/DashboardPage.vue | 13 +++++++-- frontend/src/stores/run_status.ts | 16 +++++++++++ 3 files changed, 55 insertions(+), 3 deletions(-) diff --git a/app/services/ingestion/scholar_processing.py b/app/services/ingestion/scholar_processing.py index f41d904..7867014 100644 --- a/app/services/ingestion/scholar_processing.py +++ b/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 diff --git a/frontend/src/pages/DashboardPage.vue b/frontend/src/pages/DashboardPage.vue index fbd0b4c..d5a82ee 100644 --- a/frontend/src/pages/DashboardPage.vue +++ b/frontend/src/pages/DashboardPage.vue @@ -486,9 +486,16 @@ watch(

- Started {{ formatDate(displayedLatestRun.start_dt) }}. Processed - {{ displayedLatestRun.scholar_count }} scholars and discovered - {{ displayedLatestRun.new_publication_count }} new publications. + Started {{ formatDate(displayedLatestRun.start_dt) }}. + +

, + 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; }, }, }); From 2c591a0ca33b43f08d47a8d8205dde65ecd8cd0f Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Tue, 3 Mar 2026 18:37:53 +0100 Subject: [PATCH 02/15] fix(ingestion,frontend): show live counter immediately on run start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two issues with the initial implementation: - Counter only appeared after the first SSE event arrived (scholarProgress null kept the v-else static template showing until a scholar completed) - First events were often dropped because the EventSource hadn't subscribed yet Fix: - Backend: emit an initial scholar_progress kickoff event (0/0/N) at the start of run_scholar_iteration so the frontend receives total early - Frontend: gate on isLikelyRunning alone (not scholarProgress); use optional chaining with fallbacks so "0 / … visited · 0 finished" shows immediately when a run starts, before any SSE events arrive Co-Authored-By: Claude Sonnet 4.6 --- app/services/ingestion/scholar_processing.py | 1 + frontend/src/pages/DashboardPage.vue | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/app/services/ingestion/scholar_processing.py b/app/services/ingestion/scholar_processing.py index 7867014..1735915 100644 --- a/app/services/ingestion/scholar_processing.py +++ b/app/services/ingestion/scholar_processing.py @@ -420,6 +420,7 @@ async def run_scholar_iteration( {"visited": visited, "finished": finished, "total": total}, ) + await _emit() first_pass_cstarts = await _run_first_pass( db_session, scholars=scholars, diff --git a/frontend/src/pages/DashboardPage.vue b/frontend/src/pages/DashboardPage.vue index d5a82ee..8623927 100644 --- a/frontend/src/pages/DashboardPage.vue +++ b/frontend/src/pages/DashboardPage.vue @@ -487,9 +487,9 @@ watch(

Started {{ formatDate(displayedLatestRun.start_dt) }}. -