import { useQuery, useQueryClient } from '@tanstack/react-query' import { Fragment, useCallback, useEffect, useMemo, useState } from 'react' import { useSearchParams } from 'react-router-dom' import { batchJobsStreamUrl, batchJobsFetchErrorMessage, fetchBatchJobDetail, fetchBatchJurisdictions, fetchBatchJobsDashboard, fetchFailedVideos, fetchLaunchLog, fetchLaunchStatus, launchPipeline, stopPipeline, mergeBatchIntoDashboard, mergeBatchJurisdictions, type BatchJob, type BatchJurisdictionRun, type BatchJobsDashboardPayload, type BatchVideoResult, type PipelineStage, type StageReportRow, } from '../api/batchJobs' import DeploymentPanel from '../components/DeploymentPanel' import { LinkifiedText } from '../utils/linkifiedText' import { formatCompactHours, formatCompactNumber, formatCompactPair, formatFullNumber, } from '../utils/formatCompact' import { aggregateRunningFileTiming, avgSecondsPerFile, filterJurisdictionsByState, jurisdictionStateCodes, remainingVideosForRunningBatches, resolveRunningFileClock, jurisdictionLastUpdatedIso, latestDashboardActivityIso, remainingVideosForBatch, sortJurisdictions, useTickingSeconds, } from '../utils/batchJobTiming' import { formatAgoCompact, formatDateTimeAbsolute, formatUpdatedAt, } from '../utils/dateTime' import { STATE_CODES } from '../lib/usStates' type FailedVideoRow = { batch_id: string batch_step: string state_code: string jurisdiction_id: string jurisdiction_name: string video: BatchVideoResult } const FAILED_VIDEO_STATUSES = new Set([ 'fail', 'failed', 'tombstoned', 'empty', 'rate_limit', 'error', ]) function isFailedVideoStatus(status: string): boolean { const s = (status || '').toLowerCase() if (!s || s === 'ok' || s === 'pending' || s === 'skipped') return false if (FAILED_VIDEO_STATUSES.has(s)) return true return s !== 'ok' } function failedVideoCount(j: BatchJurisdictionRun): number { const fromVideos = (j.videos || []).filter((v) => isFailedVideoStatus(v.status)).length const fromStats = Number(j.stats?.fail ?? 0) return Math.max(fromVideos, fromStats) } // Which runnable step advances each stage (used for the per-stage log drill-down). const STAGE_STEP: Record = { discover: 'discover', videos: 'catalog', transcripts: 'captions', analyses: 'analyze', reports: 'analyze', } function fmtSecs(s?: number | null): string { if (s == null || !Number.isFinite(s)) return '—' const v = Math.round(s) if (v < 60) return `${v}s` const m = Math.floor(v / 60) const r = v % 60 return r ? `${m}m ${r}s` : `${m}m` } function baseName(p?: string): string { if (!p) return '—' const parts = p.split('/') return parts[parts.length - 1] || p } function fmtEta(hours: number): string { if (!Number.isFinite(hours) || hours <= 0) return '—' if (hours < 1) return `${Math.round(hours * 60)}m` if (hours < 48) return `${Math.round(hours)}h` return `${(hours / 24).toFixed(1)}d` } function mergeDashboardFromStream( prev: BatchJobsDashboardPayload | undefined, next: BatchJobsDashboardPayload, ): BatchJobsDashboardPayload { if (!prev || next.detail !== 'summary') { return next } const prevById = new Map(prev.batches.map((b) => [b.batch_id, b])) return { ...next, batches: next.batches.map((b) => { const older = prevById.get(b.batch_id) if (older?.jurisdictions?.length) { return { ...b, jurisdictions: older.jurisdictions } } return b }), } } function collectFailedVideos( batches: BatchJob[], opts?: { batchId?: string; jurisdictionId?: string }, ): FailedVideoRow[] { const rows: FailedVideoRow[] = [] for (const batch of batches) { if (opts?.batchId && batch.batch_id !== opts.batchId) continue for (const j of batch.jurisdictions) { if (opts?.jurisdictionId && j.jurisdiction_id !== opts.jurisdictionId) continue for (const v of j.videos || []) { if (!isFailedVideoStatus(v.status)) continue rows.push({ batch_id: batch.batch_id, batch_step: batch.step, state_code: j.state_code, jurisdiction_id: j.jurisdiction_id, jurisdiction_name: j.jurisdiction_name, video: v, }) } } } return rows } function formatDuration(seconds: unknown): string { if (seconds == null || seconds === '') return '—' const total = Math.max(0, Math.floor(Number(seconds))) if (Number.isNaN(total)) return '—' if (total < 60) return `${total}s` const m = Math.floor(total / 60) const s = total % 60 if (m < 60) return `${m}m ${s}s` const h = Math.floor(m / 60) const remM = m % 60 if (h >= 24) { const compact = formatCompactNumber(h) return remM > 0 ? `${compact}h ${remM}m` : `${compact}h` } return remM > 0 ? `${h}h ${remM}m` : `${h}h` } function metricCountTitle(n: unknown, label: string): string | undefined { const full = formatFullNumber(n) const compact = formatCompactNumber(n, '') if (!full || full === compact) return undefined return `${label}: ${full}` } function formatVideoDuration(seconds: unknown): string { if (seconds == null || seconds === '') return '—' const total = Math.max(0, Math.floor(Number(seconds))) if (Number.isNaN(total) || total <= 0) return '—' const h = Math.floor(total / 3600) const m = Math.floor((total % 3600) / 60) const s = total % 60 if (h > 0) return `${h}:${String(m).padStart(2, '0')}:${String(s).padStart(2, '0')}` return `${m}:${String(s).padStart(2, '0')}` } function statusBadgeClass(status: string): string { if (status === 'completed' || status === 'ok') { return 'bg-emerald-100 text-emerald-800' } if (status === 'noop') { return 'bg-slate-100 text-slate-700' } if (status === 'failed' || status === 'fail') { return 'bg-red-100 text-red-800' } if (status === 'running') { return 'bg-sky-100 text-sky-800' } if (status === 'cancelled') { return 'bg-slate-200 text-slate-700' } if (status === 'pending') { return 'bg-amber-50 text-amber-900' } return 'bg-slate-100 text-slate-700' } function displayJurisdictionName(j: { jurisdiction_name?: string; jurisdiction_id?: string }): string { const raw = (j.jurisdiction_name || j.jurisdiction_id || '').trim() return raw.replace(/\s+(city|town|village|borough|municipality)\s*$/i, '').trim() || raw } function displayJurisdictionStatus(j: BatchJurisdictionRun): { label: string; badgeStatus: string } { const st = j.stats || {} if (Number(st.noop) > 0) return { label: 'noop', badgeStatus: 'noop' } if (Number(st.dry_run) > 0) return { label: 'dry-run', badgeStatus: 'noop' } return { label: j.status || 'pending', badgeStatus: j.status || 'pending' } } function SummaryCard({ label, value, title, onClick, active, emphasis, }: { label: string value: string | number title?: string onClick?: () => void active?: boolean emphasis?: 'danger' | 'default' }) { const content = ( <>
{label}
{value}
) const className = `rounded-lg border px-3 py-2 text-left shadow-sm transition-colors ${ active ? 'border-red-400 bg-red-50 ring-1 ring-red-300' : 'border-slate-200 bg-white hover:border-slate-300' } ${onClick ? 'cursor-pointer hover:bg-slate-50' : ''}` const tip = title || (onClick ? `View ${label}` : undefined) if (onClick) { return ( ) } return (
{content}
) } function ProgressBar({ pct }: { pct: number }) { const width = Math.min(100, Math.max(0, pct)) return (
) } function FailedVideosPanel({ rows, scopeLabel, subtitle, onClose, onSelectJurisdiction, }: { rows: FailedVideoRow[] scopeLabel: string subtitle?: string onClose: () => void onSelectJurisdiction?: (batchId: string, jurisdictionId: string) => void }) { return (

Failed videos

{scopeLabel} · {rows.length} video{rows.length === 1 ? '' : 's'} {subtitle ? ` · ${subtitle}` : ''}

{rows.length === 0 ? (

No per-video failure log yet. Failures are recorded when backfill runs with{' '} --batch-id.

) : ( {rows.map((row) => ( ))}
Jurisdiction Video Title Status Error
{row.video.video_id} {row.video.title || '—'} {row.video.status} {row.video.transcript_source ? (
{row.video.transcript_source}
) : null} 		{row.video.error ? ( ) : ( '—' )}
)}
) } function JurisdictionTable({ jurisdictions, selectedId, onSelect, onShowFailedVideos, emptyMessage, }: { jurisdictions: BatchJurisdictionRun[] selectedId: string | null onSelect: (id: string) => void onShowFailedVideos?: (jurisdictionId: string) => void emptyMessage?: string }) { if (!jurisdictions.length) { return (

{emptyMessage ?? ( <> No jurisdictions recorded yet. Run{' '} run_priority_states_last_n.sh captions. )}

) } return (
{jurisdictions.map((j) => { const st = j.stats || {} const fc = j.file_counts || {} const failCount = failedVideoCount(j) const disp = displayJurisdictionStatus(j) const selected = selectedId === j.jurisdiction_id const lastUpdated = formatUpdatedAt(jurisdictionLastUpdatedIso(j)) return ( onSelect(j.jurisdiction_id)} > ) })}
State Jurisdiction Status Videos Files Time Last checked
{j.state_code}
{displayJurisdictionName(j)}
{j.jurisdiction_id}
 {disp.label} {j.exit_code ? ` (${j.exit_code})` : ''} {j.status === 'pending' ? ( — ) : Number(st.noop) > 0 ? ( all cached ) : ( <> ok {st.ok ?? 0} ·{' '} {failCount > 0 && onShowFailedVideos ? ( ) : ( <>fail {failCount} )}{' '} · tomb {st.tombstoned ?? 0} )} {Number(st.noop) > 0 && !(fc.transcripts_disk ?? fc.transcripts) && !(fc.analysis_disk ?? fc.analysis) && !(fc.reports_disk ?? fc.reports) ? ( — ) : ( <> disk T {fc.transcripts_disk ?? fc.transcripts ?? 0} · A{' '} {fc.analysis_disk ?? fc.analysis ?? 0} · R{' '} {fc.reports_disk ?? fc.reports ?? 0} )} {formatDuration(j.elapsed_seconds)} {lastUpdated.display}
) } function VideoDrillDown({ jurisdiction, onlyFailed = false, }: { jurisdiction: BatchJurisdictionRun onlyFailed?: boolean }) { const videos = useMemo(() => { const all = jurisdiction.videos || [] if (!onlyFailed) return all return all.filter((v) => isFailedVideoStatus(v.status)) }, [jurisdiction.videos, onlyFailed]) return (

{onlyFailed ? 'Failed videos · ' : ''} {displayJurisdictionName(jurisdiction)}

{jurisdiction.jurisdiction_id}

{onlyFailed ? (

{videos.length} failed video{videos.length === 1 ? '' : 's'} in this batch run

) : null}
{videos.length === 0 ? ( ) : ( videos.map((v) => ( )) )}
Video Title Status Duration Source Error
{onlyFailed ? 'No failed videos logged for this jurisdiction in this batch run.' : ( <> No per-video log for this run (re-run backfill with{' '} --batch-id). )}
{v.video_id} {v.title || '—'} {v.status} {formatVideoDuration(v.duration_seconds)} {v.transcript_source || '—'} {v.error ? ( ) : ( '—' )}
) } function BatchDetailPanel({ batch, selectedJurisdictionId, onSelectJurisdiction, stateFilter, onStateFilterChange, showFailedVideos, onToggleFailedVideos, onShowFailedForJurisdiction, }: { batch: BatchJob selectedJurisdictionId: string | null onSelectJurisdiction: (id: string | null) => void stateFilter: string onStateFilterChange: (state: string) => void showFailedVideos: boolean onToggleFailedVideos: () => void onShowFailedForJurisdiction: (jurisdictionId: string) => void }) { const s = batch.summary || {} const total = Number(s.total_jurisdictions) || 0 const processed = Number(s.processed_jurisdictions) || 0 const pct = total > 0 ? Math.round((100 * processed) / total) : 0 const cfg = batch.config || {} const states = Array.isArray(cfg.states) ? (cfg.states as string[]).join(', ') : '—' const selectedJurisdiction = useMemo( () => batch.jurisdictions.find((j) => j.jurisdiction_id === selectedJurisdictionId) ?? null, [batch.jurisdictions, selectedJurisdictionId], ) const videosFailCount = Number(s.videos_fail) || 0 const isRunning = batch.status === 'running' const fileClock = useMemo(() => resolveRunningFileClock(batch), [batch]) const currentFileSeconds = useTickingSeconds( fileClock?.startedAt, isRunning && !!fileClock?.startedAt, ) const avgPerFileSec = avgSecondsPerFile(s) const batchRemainingVideos = useMemo(() => remainingVideosForBatch(batch), [batch]) const stateCodes = useMemo(() => { const fromCfg = Array.isArray(cfg.states) ? (cfg.states as string[]).map((s) => String(s).toUpperCase().trim()).filter(Boolean) : [] if (fromCfg.length) return [...new Set(fromCfg)].sort() return jurisdictionStateCodes(batch.jurisdictions) }, [batch.jurisdictions, cfg.states]) const sortedJurisdictions = useMemo( () => sortJurisdictions(batch.jurisdictions), [batch.jurisdictions], ) const displayedJurisdictions = useMemo( () => filterJurisdictionsByState(sortedJurisdictions, stateFilter), [sortedJurisdictions, stateFilter], ) const showJurisdictionTable = !!stateFilter return (

{batch.step}

{batch.status}

{batch.batch_id}

Started {formatDateTimeAbsolute(batch.started_at)} {batch.finished_at ? ` · Finished ${formatDateTimeAbsolute(batch.finished_at)}` : ''}

{isRunning && fileClock?.videoId ? (

Current:{' '} {fileClock.videoId} {fileClock.title ? ` · ${fileClock.title.slice(0, 72)}` : ''}

) : null}
{isRunning ? ( <> ) : null} 0 ? 'danger' : 'default'} active={showFailedVideos} onClick={videosFailCount > 0 ? onToggleFailedVideos : undefined} />

States: {states} · N={String(cfg.n ?? '—')} · delay={String(cfg.delay ?? '—')}s · source= {String(cfg.transcript_source || '—')}

Jurisdictions {showJurisdictionTable ? ( ({displayedJurisdictions.length} in {stateFilter}) ) : null}

{showJurisdictionTable ? ( onSelectJurisdiction(id)} onShowFailedVideos={onShowFailedForJurisdiction} emptyMessage={`No jurisdictions in ${stateFilter} for this batch.`} /> ) : (

{selectedJurisdictionId ? 'Per-jurisdiction details are below. Pick a state to browse the full list for that state.' : 'Select a state to browse jurisdictions. The full plan is large; filter by state to keep the table fast.'}

)}
{selectedJurisdiction ? ( ) : null}
) } // Priority dev states — mirrors the wrapper-script default (the STATES env in // youtube_run_priority_states_last_n.sh). "All states" uses STATE_CODES (50 + DC). const PRIORITY_STATE_CODES = ['AL', 'GA', 'IN', 'MA', 'MT', 'WA', 'WI'] type JobType = 'youtube' | 'deployment' function JobTypeTabs({ active, onChange, }: { active: JobType onChange: (t: JobType) => void }) { const tabs: { key: JobType; label: string }[] = [ { key: 'youtube', label: 'YouTube pipeline' }, { key: 'deployment', label: 'Prod deployment' }, ] return (
{tabs.map((t) => ( ))}
) } export default function BatchJobStatusPage() { const [searchParams, setSearchParams] = useSearchParams() const jobType: JobType = searchParams.get('type') === 'deployment' ? 'deployment' : 'youtube' const setJobType = useCallback( (t: JobType) => { const next = new URLSearchParams(searchParams) if (t === 'deployment') next.set('type', 'deployment') else next.delete('type') setSearchParams(next, { replace: true }) }, [searchParams, setSearchParams], ) const batchId = searchParams.get('batch') ?? '' const jurisdictionId = searchParams.get('jurisdiction') ?? '' const stateFilter = (searchParams.get('state') ?? '').toUpperCase() const showFailedVideos = searchParams.get('view') === 'failed-videos' const failedVideosScopeAll = searchParams.get('failed_scope') === 'all' const queryClient = useQueryClient() const [streamLive, setStreamLive] = useState(false) const [videosLoading, setVideosLoading] = useState(false) const { data, isPending, isFetching, isError, error, refetch } = useQuery({ queryKey: ['batch-jobs-dashboard'], queryFn: () => fetchBatchJobsDashboard(false, 'summary'), staleTime: 30_000, refetchOnWindowFocus: true, retry: 1, }) useEffect(() => { const url = batchJobsStreamUrl() const es = new EventSource(url) const applyPayload = (payload: BatchJobsDashboardPayload) => { queryClient.setQueryData(['batch-jobs-dashboard'], (prev) => mergeDashboardFromStream(prev, payload), ) } es.onopen = () => setStreamLive(true) es.addEventListener('summary', (ev) => { try { applyPayload(JSON.parse(ev.data) as BatchJobsDashboardPayload) } catch { /* ignore */ } }) es.onerror = () => { setStreamLive(false) es.close() void refetch() } return () => { setStreamLive(false) es.close() } }, [queryClient, refetch]) const { data: launchStatus, refetch: refetchLaunch } = useQuery({ queryKey: ['batch-launch-status'], queryFn: fetchLaunchStatus, refetchInterval: 20_000, retry: false, }) const [launching, setLaunching] = useState(false) const [stopping, setStopping] = useState(false) const [launchMsg, setLaunchMsg] = useState(null) // Parallelism = how many jurisdictions analyze at once (the main throughput // lever). Persisted so it survives reloads; ceiling is the Gemini quota. const [parallelism, setParallelism] = useState(() => { const v = Number(localStorage.getItem('batch-parallelism')) return v >= 1 && v <= 8 ? v : 6 }) const setParallel = useCallback((v: number) => { setParallelism(v) localStorage.setItem('batch-parallelism', String(v)) }, []) // Which states a full / "ALL"-scope launch targets: the priority dev set // (matches the script default) or every US state + DC. Picking a single state // from the scope dropdown overrides this. Persisted across reloads. const [allScope, setAllScopeState] = useState<'priority' | 'all'>(() => { return localStorage.getItem('batch-launch-scope') === 'all' ? 'all' : 'priority' }) const setAllScope = useCallback((v: 'priority' | 'all') => { setAllScopeState(v) localStorage.setItem('batch-launch-scope', v) }, []) // How many missing transcripts one ▶ Backfill click fetches. 0 = "All" (sweep // every missing transcript). The run is resumable — it re-queries what's still // missing — so a bounded size still makes incremental progress across clicks. // Persisted across reloads; defaults to 4000. const BACKFILL_SIZES = [100, 500, 2000, 4000, 0] as const const [backfillSize, setBackfillSizeState] = useState(() => { const raw = localStorage.getItem('batch-backfill-size') const v = Number(raw) return raw != null && (BACKFILL_SIZES as readonly number[]).includes(v) ? v : 4000 }) const setBackfillSize = useCallback((v: number) => { setBackfillSizeState(v) localStorage.setItem('batch-backfill-size', String(v)) }, []) const runLaunch = useCallback( async (step: string, states: string[], nOverride?: number) => { setLaunching(true) setLaunchMsg(null) try { const res = await launchPipeline({ step, states, parallel: parallelism, n: nOverride ?? 25, }) setLaunchMsg(res.detail || 'Launched.') void refetchLaunch() window.setTimeout(() => void refetch(), 2500) } catch (e) { setLaunchMsg((e as Error).message || 'Launch failed') } finally { setLaunching(false) } }, [refetchLaunch, refetch, parallelism], ) const stopLaunch = useCallback( async (step?: string) => { const label = step ? `the '${step}' run` : 'all running jobs' if (!window.confirm(`Stop ${label}? In-flight work checkpoints and the process exits.`)) { return } setStopping(true) setLaunchMsg(null) try { const res = await stopPipeline(step ? { step } : {}) setLaunchMsg(res.detail || 'Stop requested.') void refetchLaunch() window.setTimeout(() => void refetch(), 2500) } catch (e) { setLaunchMsg((e as Error).message || 'Stop failed') } finally { setStopping(false) } }, [refetchLaunch, refetch], ) // Per-stage drill-down: expand one stage to see its live log + current file. const [expandedStage, setExpandedStage] = useState(null) // The Transcripts stage is advanced by two steps — per-jurisdiction `captions` // and the global `backfill` — which write separate launch logs. Show the // backfill log while it's the live (running/stalled) step; otherwise the // conventional captions log. const transcriptsBackfillLive = (launchStatus?.running_steps ?? []).includes('backfill') || (launchStatus?.stalled_steps ?? []).includes('backfill') const expandedStep = expandedStage ? expandedStage === 'transcripts' && transcriptsBackfillLive ? 'backfill' : STAGE_STEP[expandedStage] : null const { data: stageLog } = useQuery({ queryKey: ['launch-log', expandedStep], queryFn: () => fetchLaunchLog(expandedStep as string), enabled: !!expandedStep, refetchInterval: 4000, retry: false, }) const selectedBatch = useMemo(() => { if (!data?.batches?.length) return null if (batchId) { return data.batches.find((b) => b.batch_id === batchId) ?? data.batches[0] } return data.batches[0] }, [data, batchId]) const setBatch = useCallback( (id: string) => { const next = new URLSearchParams(searchParams) next.set('batch', id) next.delete('jurisdiction') next.delete('state') setSearchParams(next, { replace: true }) }, [searchParams, setSearchParams], ) const setStateFilter = useCallback( (state: string) => { const next = new URLSearchParams(searchParams) if (selectedBatch?.batch_id) next.set('batch', selectedBatch.batch_id) const st = state.trim().toUpperCase() if (st) next.set('state', st) else next.delete('state') if (jurisdictionId && selectedBatch) { const visible = filterJurisdictionsByState( sortJurisdictions(selectedBatch.jurisdictions), st, ) if (!visible.some((j) => j.jurisdiction_id === jurisdictionId)) { next.delete('jurisdiction') } } setSearchParams(next, { replace: true }) }, [searchParams, setSearchParams, selectedBatch, jurisdictionId], ) const setJurisdiction = useCallback( (id: string | null) => { const next = new URLSearchParams(searchParams) if (selectedBatch?.batch_id) next.set('batch', selectedBatch.batch_id) if (id) next.set('jurisdiction', id) else next.delete('jurisdiction') next.delete('view') setSearchParams(next, { replace: true }) }, [searchParams, setSearchParams, selectedBatch?.batch_id], ) const setShowFailedVideos = useCallback( ( open: boolean, opts?: { batchId?: string; jurisdictionId?: string; allBatches?: boolean }, ) => { const next = new URLSearchParams(searchParams) if (opts?.batchId) next.set('batch', opts.batchId) if (opts?.jurisdictionId) next.set('jurisdiction', opts.jurisdictionId) else if (!open) next.delete('jurisdiction') if (open) { next.set('view', 'failed-videos') if (opts?.allBatches) next.set('failed_scope', 'all') else next.delete('failed_scope') } else { next.delete('view') next.delete('failed_scope') } setSearchParams(next, { replace: true }) }, [searchParams, setSearchParams], ) const planStateCodes = useMemo(() => { if (!selectedBatch) return [] as string[] const fromCfg = Array.isArray(selectedBatch.config?.states) ? (selectedBatch.config.states as string[]) .map((s) => String(s).toUpperCase().trim()) .filter(Boolean) : [] if (fromCfg.length) return [...new Set(fromCfg)].sort() return jurisdictionStateCodes(selectedBatch.jurisdictions) }, [selectedBatch]) const effectiveStateFilter = useMemo(() => { if (!selectedBatch || !stateFilter) return '' return planStateCodes.includes(stateFilter) ? stateFilter : '' }, [selectedBatch, stateFilter, planStateCodes]) const { data: stateJurisdictions, isFetching: stateJurisdictionsFetching, } = useQuery({ queryKey: ['batch-jurisdictions', selectedBatch?.batch_id, effectiveStateFilter], queryFn: () => fetchBatchJurisdictions(selectedBatch!.batch_id, effectiveStateFilter), enabled: !!selectedBatch?.batch_id && !!effectiveStateFilter, staleTime: 15_000, }) const selectedBatchForPanel = useMemo(() => { if (!selectedBatch) return null if (!effectiveStateFilter) return selectedBatch if (stateJurisdictions) { return mergeBatchJurisdictions(selectedBatch, stateJurisdictions) } return selectedBatch }, [selectedBatch, effectiveStateFilter, stateJurisdictions]) const visibleJurisdictions = useMemo(() => { if (!selectedBatchForPanel || !effectiveStateFilter) return [] return filterJurisdictionsByState( sortJurisdictions(selectedBatchForPanel.jurisdictions), effectiveStateFilter, ) }, [selectedBatchForPanel, effectiveStateFilter]) const effectiveJurisdictionId = useMemo(() => { if (!jurisdictionId || !selectedBatch) return null if (!effectiveStateFilter) return jurisdictionId return visibleJurisdictions.some((j) => j.jurisdiction_id === jurisdictionId) ? jurisdictionId : null }, [jurisdictionId, selectedBatch, effectiveStateFilter, visibleJurisdictions]) const selectedJurisdictionForVideos = useMemo(() => { if (!selectedBatchForPanel || !effectiveJurisdictionId) return null return ( selectedBatchForPanel.jurisdictions.find( (j) => j.jurisdiction_id === effectiveJurisdictionId, ) ?? null ) }, [selectedBatchForPanel, effectiveJurisdictionId]) const failedVideosBatchId = failedVideosScopeAll ? undefined : selectedBatch?.batch_id const { data: failedVideosPayload, isFetching: failedVideosFetching, } = useQuery({ queryKey: ['batch-failed-videos', failedVideosBatchId ?? 'all'], queryFn: () => fetchFailedVideos(failedVideosBatchId), enabled: showFailedVideos && !effectiveJurisdictionId && (data?.totals.videos_fail ?? 0) > 0, staleTime: 30_000, }) useEffect(() => { if (!selectedBatch?.batch_id) return const needsJurisdictionVideos = !!effectiveJurisdictionId && !!selectedJurisdictionForVideos && (showFailedVideos ? failedVideoCount(selectedJurisdictionForVideos) > 0 && !(selectedJurisdictionForVideos.videos || []).some((v) => isFailedVideoStatus(v.status), ) : (selectedJurisdictionForVideos.videos?.length ?? 0) === 0 && (Number(selectedJurisdictionForVideos.stats?.ok ?? 0) > 0 || failedVideoCount(selectedJurisdictionForVideos) > 0)) if (!needsJurisdictionVideos) return let cancelled = false setVideosLoading(true) const mode = 'all' void fetchBatchJobDetail(selectedBatch.batch_id, mode) .then((batch) => { if (cancelled) return queryClient.setQueryData( ['batch-jobs-dashboard'], (prev) => mergeBatchIntoDashboard(prev, batch) ?? prev, ) }) .finally(() => { if (!cancelled) setVideosLoading(false) }) return () => { cancelled = true } }, [ selectedBatch?.batch_id, effectiveJurisdictionId, showFailedVideos, selectedJurisdictionForVideos, data?.totals.videos_fail, queryClient, ]) const allFailedVideoRows = useMemo((): FailedVideoRow[] => { if (effectiveJurisdictionId) { return collectFailedVideos(data?.batches ?? [], { batchId: selectedBatch?.batch_id, jurisdictionId: effectiveJurisdictionId, }) } if (failedVideosPayload?.rows?.length) { return failedVideosPayload.rows.map((r) => ({ batch_id: r.batch_id, batch_step: r.batch_step, state_code: r.state_code, jurisdiction_id: r.jurisdiction_id, jurisdiction_name: r.jurisdiction_name, video: r.video, })) } return collectFailedVideos(data?.batches ?? [], { batchId: failedVideosBatchId, }) }, [ data?.batches, selectedBatch?.batch_id, effectiveJurisdictionId, failedVideosPayload, failedVideosBatchId, ]) const failedScopeLabel = useMemo(() => { if (effectiveJurisdictionId && selectedBatch) { const j = selectedBatch.jurisdictions.find( (x) => x.jurisdiction_id === effectiveJurisdictionId, ) return `${selectedBatch.step} · ${j?.jurisdiction_name || effectiveJurisdictionId}` } if (failedVideosScopeAll) return 'All recent batches' if (selectedBatch) return `Batch ${selectedBatch.step} (${selectedBatch.batch_id})` return 'All batches' }, [effectiveJurisdictionId, selectedBatch, failedVideosScopeAll]) const failedVideosSubtitle = useMemo(() => { if (effectiveJurisdictionId) return undefined const total = failedVideosPayload?.total_fail_in_summaries ?? data?.totals.videos_fail ?? 0 const parts: string[] = [] if (total > allFailedVideoRows.length) { parts.push(`${total} in summaries`) } if (failedVideosPayload?.truncated) { parts.push('list capped at 500') } return parts.length ? parts.join('; ') : undefined }, [ effectiveJurisdictionId, failedVideosPayload, data?.totals.videos_fail, allFailedVideoRows.length, ]) const lastActivityIso = useMemo(() => { const fromApi = data?.last_activity_at?.trim() if (fromApi) return fromApi const fromBatches = latestDashboardActivityIso(data?.batches ?? []) if (fromBatches) return fromBatches const running = data?.batches?.find((b) => b.status === 'running') return running?.updated_at?.trim() || null }, [data?.last_activity_at, data?.batches]) const [agoClockMs, setAgoClockMs] = useState(() => Date.now()) useEffect(() => { const id = window.setInterval(() => setAgoClockMs(Date.now()), 10_000) return () => window.clearInterval(id) }, []) const lastUpdateAgo = useMemo( () => (lastActivityIso ? formatAgoCompact(lastActivityIso, agoClockMs) : null), [lastActivityIso, agoClockMs], ) const lastUpdateAbsolute = useMemo( () => (lastActivityIso ? formatUpdatedAt(lastActivityIso) : null), [lastActivityIso], ) // Analyze/report runs don't touch the batch tracker (see "Last batch step"), so // these "ago" cards read the most recent policy stamps instead — the real signal // that the analyze pipeline is alive, including standalone/parallel runs. const lastTranscriptIso = data?.totals.last_transcript_at?.trim() || null const lastAnalysisIso = data?.totals.last_analysis_at?.trim() || null const lastReportIso = data?.totals.last_report_at?.trim() || null const lastTranscriptAgo = useMemo( () => (lastTranscriptIso ? formatAgoCompact(lastTranscriptIso, agoClockMs) : null), [lastTranscriptIso, agoClockMs], ) const lastTranscriptAbsolute = useMemo( () => (lastTranscriptIso ? formatUpdatedAt(lastTranscriptIso) : null), [lastTranscriptIso], ) const lastAnalysisAgo = useMemo( () => (lastAnalysisIso ? formatAgoCompact(lastAnalysisIso, agoClockMs) : null), [lastAnalysisIso, agoClockMs], ) const lastReportAgo = useMemo( () => (lastReportIso ? formatAgoCompact(lastReportIso, agoClockMs) : null), [lastReportIso, agoClockMs], ) const lastAnalysisAbsolute = useMemo( () => (lastAnalysisIso ? formatUpdatedAt(lastAnalysisIso) : null), [lastAnalysisIso], ) const lastReportAbsolute = useMemo( () => (lastReportIso ? formatUpdatedAt(lastReportIso) : null), [lastReportIso], ) const runningTiming = useMemo( () => aggregateRunningFileTiming(data?.batches ?? []), [data?.batches], ) const remainingVideos = useMemo( () => remainingVideosForRunningBatches(data?.batches ?? []), [data?.batches], ) const globalCurrentFileSeconds = useTickingSeconds( runningTiming.activeVideo?.startedAt, (data?.totals.running ?? 0) > 0 && !!runningTiming.activeVideo, ) if (jobType === 'deployment') { return (
) } return (

Live from bronze.youtube_batch_job_runs via{' '} GET /api/batch-jobs/stream {streamLive && ( connected )}

{isPending && !data && (
Loading batch summary…
)} {isFetching && data && (

Refreshing batch summary…

)} {isError && (
Could not load batch jobs: {batchJobsFetchErrorMessage(error)}

Start or restart the API:{' '} uvicorn api.main:app --reload --port 8001 . Apply migration 073 if needed. Batch runs sync from{' '} run_priority_states_last_n.sh. {!streamLive && ( <> {' '} )}

)} {data && ( <> {(() => { const report = data.stage_report ?? { states: [], rows: [] } const stateOpts = report.states ?? [] const scope = stateFilter && stateOpts.includes(stateFilter) ? stateFilter : 'ALL' const rowFor = ( sc: string, stage: PipelineStage, ): StageReportRow | undefined => report.rows.find((r) => r.scope === sc && r.stage === stage) const fmtPct = (done: number, total: number) => { if (!total) return '—' const r = Math.round((done / total) * 1000) / 10 return `${Number.isInteger(r) ? r : r.toFixed(1)}%` } const agoFromIso = (iso?: string) => iso ? formatAgoCompact(iso, agoClockMs) : null const openFailed = () => setShowFailedVideos(true, { allBatches: true }) const setScope = (sc: string) => { const next = new URLSearchParams(searchParams) if (sc) next.set('state', sc) else next.delete('state') setSearchParams(next, { replace: true }) } const stageList: PipelineStage[] = [ 'discover', 'videos', 'transcripts', 'analyses', 'reports', ] const stageLabel: Record = { discover: 'Discover', videos: 'Videos', transcripts: 'Transcripts', analyses: 'Analyses', reports: 'Reports', } // Discover-stage per-entity split labels (counties vs municipalities). const entityShort: Record = { counties: 'Co', municipalities: 'Mun', } const entityLabel: Record = { counties: 'Counties', municipalities: 'Municipalities (cities/towns)', } // Which runnable step each stage's "Run" button kicks off. const stageStep: Record = { discover: 'discover', videos: 'catalog', transcripts: 'captions', analyses: 'analyze', reports: 'analyze', } const stepDesc: Record = { discover: 'discover channels (scrape sites / search YouTube)', videos: 'catalog — list videos on the channel', transcripts: 'captions — download transcripts', analyses: 'analyze — Gemini policy analysis', reports: 'analyze — regenerates reports too', } type Def = { n: number stage: PipelineStage unit: string sub?: string pill: boolean drill: (() => void) | null failTitle: string } const defs: Def[] = [ { n: 0, stage: 'discover', unit: 'jurisdictions', sub: 'channel discovery — scrape county sites / search YouTube', pill: true, drill: null, failTitle: 'Jurisdictions still missing a YouTube channel', }, { n: 1, stage: 'videos', unit: 'attempted', pill: false, drill: openFailed, failTitle: 'Videos whose caption fetch failed — tap to list', }, { n: 2, stage: 'transcripts', unit: 'videos', sub: `${formatCompactNumber(data.totals.files_transcripts_disk ?? 0)} segment rows · deduped to videos`, pill: true, drill: openFailed, failTitle: 'Videos with no usable transcript — tap to list', }, { n: 3, stage: 'analyses', unit: 'videos', pill: true, drill: null, failTitle: 'Analysis errors (bronze policy_analysis_error)', }, { n: 4, stage: 'reports', unit: 'videos', pill: true, drill: null, failTitle: 'Report errors (bronze policy_report_error)', }, ] // Fixed-width Failed/Run columns so the progress column lines up across rows. const cols = 'grid grid-cols-[1.4fr_0.8fr_minmax(0,1.7fr)_4.5rem_6.5rem] items-center gap-3' const ls = launchStatus // ALL view scope → an explicit launch target (priority set or 50 + DC) // so the backend can't silently fall back to its priority-only default. const launchScopeStates = scope === 'ALL' ? allScope === 'all' ? STATE_CODES : PRIORITY_STATE_CODES : [scope] const scopeLaunchLabel = scope !== 'ALL' ? scope : allScope === 'all' ? 'all 50 + DC' : 'priority' // Each running step maps to the stage(s) it advances. Different steps // run concurrently, so we union them for the live "running" markers. const stepStages: Record = { discover: ['discover'], catalog: ['videos'], captions: ['videos', 'transcripts'], // Global mart-wide sweep — advances only the transcripts stage. backfill: ['transcripts'], analyze: ['analyses', 'reports'], all: ['discover', 'videos', 'transcripts', 'analyses', 'reports'], each: ['discover', 'videos', 'transcripts', 'analyses', 'reports'], } const runningSteps = new Set(ls?.running_steps ?? []) const stalledSteps = new Set(ls?.stalled_steps ?? []) const runningStages = new Set() runningSteps.forEach((s) => (stepStages[s] ?? []).forEach((g) => runningStages.add(g)), ) const anyFullRun = runningSteps.has('all') || runningSteps.has('each') // A stage's Run is allowed unless its step (or a full-pipeline run) is // already running — so e.g. Discover can launch while Analyze runs. const canRunStage = (g: PipelineStage) => !!ls?.enabled && !launching && !runningSteps.has(stageStep[g]) && !anyFullRun // The global backfill is its own step, so it may run alongside // captions; it's only blocked by another backfill or a full run. const canRunBackfill = !!ls?.enabled && !launching && !runningSteps.has('backfill') && !anyFullRun const canRunAll = !!ls?.enabled && !launching && runningSteps.size === 0 && data.totals.running === 0 // Header "last activity" = the freshest per-step stamp for this scope, // not the stale batch-step clock. const freshestIso = stageList .map((s) => rowFor(scope, s)?.last_at || '') .reduce((a, b) => (a > b ? a : b), '') const freshestAgo = agoFromIso(freshestIso) // Mirrors the API stall window (_LAUNCH_STALL_SECONDS / // BATCH_JOB_INACTIVITY_SECONDS, default 3600s). const STALL_SECONDS = 3600 const freshestAgeSecs = freshestIso ? Math.max(0, Math.floor((agoClockMs - Date.parse(freshestIso)) / 1000)) : null // A batch row stays status='running' after its worker dies or starts // skipping already-done work, so totals.running alone overstates // liveness. Count row-running as live only with recent recorded // progress; otherwise it's a stalled run the reaper hasn't swept yet. const rowsStale = data.totals.running > 0 && runningSteps.size === 0 && freshestAgeSecs != null && freshestAgeSecs > STALL_SECONDS const liveRunningRows = data.totals.running > 0 && !rowsStale const isActive = liveRunningRows || runningSteps.size > 0 const anyStalled = stalledSteps.size > 0 || rowsStale const statusText = isActive ? 'Running' : anyStalled ? 'Stalled' : 'Idle' const statusDot = isActive ? 'bg-emerald-500' : anyStalled ? 'bg-amber-500' : 'bg-slate-400' return ( <>
{statusText} {freshestAgo ? ( · last activity {freshestAgo} ) : null} {runningSteps.size > 0 ? `· running ${[...runningSteps].join(', ')}` : rowsStale ? `· ${formatCompactNumber(data.totals.running)} running, no progress >1h — likely stalled` : stalledSteps.size > 0 ? `· ${[...stalledSteps].join(', ')} timed out (no activity 1h)` : `· ${formatCompactNumber(data.totals.running)} running`} {scope !== 'ALL' ? ( {scope} ) : null}
{ls?.enabled && scope === 'ALL' ? ( ) : null} {ls?.enabled ? ( ) : null} {ls?.enabled ? ( ) : null} {ls?.enabled && (ls?.busy || runningSteps.size > 0 || stalledSteps.size > 0) ? ( ) : null} {stateOpts.length > 0 ? ( ) : null}
{launchMsg ? (
{launchMsg}
) : null}
Stage
Last activity
Progress
Failed
Run
{defs.map((st, idx) => { const r = rowFor(scope, st.stage) const done = r?.done ?? 0 const total = r?.total ?? 0 const failed = r?.failed ?? 0 const pct = total ? (done / total) * 100 : 0 const ago = agoFromIso(r?.last_at) const running = runningStages.has(st.stage) const expanded = expandedStage === st.stage const tm = data.stage_report?.timing?.[st.stage] // Live pace: avg_seconds is the median gap between *completed* // files, so it freezes during a stall (no new gap is ever // recorded). Once a running stage is overdue for its next // completion, our best estimate of the current pace is at // least how long we've already been waiting — fold the open // trailing gap in so /hr and ETA degrade every clock tick and // recover the moment a file lands. const avgSecs = tm?.avg_seconds && tm.avg_seconds > 0 ? tm.avg_seconds : null const lastAtMs = tm?.last_at ? Date.parse(tm.last_at) : NaN const sinceLastSecs = Number.isFinite(lastAtMs) ? Math.max(0, (agoClockMs - lastAtMs) / 1000) : null const effectiveSecs = avgSecs == null ? null : running && sinceLastSecs != null ? Math.max(avgSecs, sinceLastSecs) : avgSecs const stalled = running && avgSecs != null && sinceLastSecs != null && sinceLastSecs > avgSecs * 3 return (
{running ? ( <> running… ) : ( <> {ago ?? '—'} )}
{formatCompactNumber(done)} {' '} / {formatCompactNumber(total)} {st.unit} · {fmtPct(done, total)}
{st.stage === 'discover' && (r?.breakdown?.length ?? 0) > 0 ? (
{r!.breakdown!.map((b) => ( {entityLabel[b.entity] ?? b.entity} {' '} {fmtPct(b.done, b.total)}{' '} {formatCompactNumber(b.done)}/{formatCompactNumber(b.total)} ))}
) : null}
{failed > 0 ? ( st.drill ? ( ) : ( {formatCompactNumber(failed)} ) ) : ( 0 )}
{ls?.enabled ? ( ) : null} {/* Transcripts only: global mart-wide sweep (also reaches LocalView/union videos captions never visits). The size picker sets how many missing transcripts one click fetches (All = sweep everything). */} {ls?.enabled && st.stage === 'transcripts' ? ( <> ) : null}
{expanded ? (
avg {fmtSecs(avgSecs)}/file {effectiveSecs ? ` · ${Math.round(3600 / effectiveSecs)}/hr` : ''} {stalled ? ( · stalled ) : null} {effectiveSecs && total - done > 0 ? ( ETA{' '} {fmtEta(((total - done) * effectiveSecs) / 3600)} {' '} for {formatCompactNumber(total - done)} left ) : null} last file:{' '} {baseName(tm?.last_path)} {tm?.last_at ? ` · ${agoFromIso(tm.last_at) ?? ''}` : ''} {running && stageLog?.current ? ( current:{' '} {stageLog.current} {stageLog.current_since ? ` · ${agoFromIso(stageLog.current_since) ?? ''} on it` : ''} ) : null} 
                          {stageLog?.lines?.length
                            ? stageLog.lines.join('\n')
                            : 'No log for this step yet — launch it to see live output.'}
{stageLog?.path ? (
log: {stageLog.path}
) : null}
) : null} ) })}
● {' '} retryable{' '} ▪ {' '} terminal · tap a failed count to drill in {ls && !ls.enabled ? ' · launch disabled (set BATCH_JOBS_ALLOW_LAUNCH=1)' : ''}
{stateOpts.length > 0 ? (
State coverage · {stateOpts.length} states · click a row to scope
{stageList.map((s) => ( ))} {stateOpts.map((stCode) => ( setScope(stCode)} className={`cursor-pointer border-t border-slate-100 hover:bg-slate-50 ${ scope === stCode ? 'bg-slate-50' : '' }`} > {stageList.map((s) => { const r = rowFor(stCode, s) const done = r?.done ?? 0 const total = r?.total ?? 0 const bd = s === 'discover' ? r?.breakdown ?? [] : [] return ( ) })} ))}
State {stageLabel[s]}
{stCode}
{fmtPct(done, total)}
{formatCompactNumber(done)}/{formatCompactNumber(total)}
{bd.length > 0 ? (
{bd.map((b) => (
{entityShort[b.entity] ?? b.entity} {' '} {fmtPct(b.done, b.total)}{' '} {formatCompactNumber(b.done)}/{formatCompactNumber(b.total)}
))}
) : null}
) : null} ) })()}
All metrics
{lastActivityIso && lastUpdateAgo ? ( ) : null} {lastTranscriptIso && lastTranscriptAgo ? ( ) : null} {lastAnalysisIso && lastAnalysisAgo ? ( ) : null} {lastReportIso && lastReportAgo ? ( ) : null} 0 ? `${formatFullNumber(data.totals.running)} batch(es) with status running; ${runningTiming.idleRunningBatchCount} idle (no active file). Idle runs auto-cancel after 1h without progress.` : metricCountTitle(data.totals.running, 'Running batches') } /> {data.totals.running > 0 ? ( <> 0 ? `${runningTiming.idleRunningBatchCount} running batch(es) with no in-flight video (between jurisdictions or waiting).` : 'No in-flight video with a start time' } /> ) : null} 0 ? 'danger' : 'default'} active={showFailedVideos} onClick={ data.totals.videos_fail > 0 ? () => setShowFailedVideos(!showFailedVideos, { allBatches: !showFailedVideos, }) : undefined } /> 0 ? `${Math.round((data.totals.files_analysis / data.totals.files_transcripts_disk) * 100)}%` : '—' } title={`${formatCompactNumber(data.totals.files_analysis)} of ${formatCompactNumber( data.totals.files_transcripts_disk ?? 0, )} transcripts have an AI analysis (live bronze stamps)`} /> 0 ? `${Math.round((data.totals.files_reports / data.totals.files_transcripts_disk) * 100)}%` : '—' } title={`${formatCompactNumber(data.totals.files_reports)} of ${formatCompactNumber( data.totals.files_transcripts_disk ?? 0, )} transcripts have a generated report (live bronze stamps; same denominator as Analysis progress, so the two read as one funnel — not reports÷analyses)`} />
{showFailedVideos && !effectiveJurisdictionId && ( setShowFailedVideos(false)} onSelectJurisdiction={(bid, jid) => { setShowFailedVideos(true, { batchId: bid, jurisdictionId: jid }) }} /> )}
{videosLoading || stateJurisdictionsFetching ? (

{videosLoading ? 'Loading per-video details…' : 'Loading jurisdictions for selected state…'}

) : null}
{selectedBatchForPanel ? ( setShowFailedVideos(!showFailedVideos, { batchId: selectedBatchForPanel.batch_id, allBatches: false, }) } onShowFailedForJurisdiction={(jid) => { setShowFailedVideos(true, { batchId: selectedBatchForPanel.batch_id, jurisdictionId: jid, }) requestAnimationFrame(() => { document .getElementById('batch-jurisdiction-drilldown') ?.scrollIntoView({ behavior: 'smooth', block: 'nearest' }) }) }} /> ) : (
Select a batch
)}
)}
) }