diff --git a/web-studio/src/routes/monitoring/-components/queue-status-card.tsx b/web-studio/src/routes/monitoring/-components/queue-status-card.tsx new file mode 100644 index 0000000000..ed078eb60b --- /dev/null +++ b/web-studio/src/routes/monitoring/-components/queue-status-card.tsx @@ -0,0 +1,172 @@ +import * as React from 'react' +import { useTranslation } from 'react-i18next' +import { Card, CardTitle } from '#/components/ui/card' +import { cn } from '#/lib/utils' +import { parseObserverStatus } from '../-lib/parse-status' + +export interface ParsedQueueRow { + name: string + processing: number + pending: number + completed: number + errors: number + total: number +} + +// 将 Observer status 字符串解析为结构化队列数据 +function parseQueueStatus(status: string): ParsedQueueRow[] { + if (!status) return [] + const blocks = parseObserverStatus(status) + const tableBlock = blocks.find((b) => b.kind === 'table') + if (!tableBlock) return [] + + const headers = tableBlock.headers + const col = { + name: headers.findIndex((h) => h === 'Queue'), + inProgress: headers.findIndex((h) => h === 'In Progress'), + pending: headers.findIndex((h) => h === 'Pending'), + processed: headers.findIndex((h) => h === 'Processed'), + errors: headers.findIndex((h) => h === 'Errors'), + total: headers.findIndex((h) => h === 'Total'), + } + + return tableBlock.rows.map((row) => ({ + name: col.name >= 0 ? (row[col.name] ?? '') : '', + processing: col.inProgress >= 0 ? (parseInt(row[col.inProgress] ?? '0', 10) || 0) : 0, + pending: col.pending >= 0 ? (parseInt(row[col.pending] ?? '0', 10) || 0) : 0, + completed: col.processed >= 0 ? (parseInt(row[col.processed] ?? '0', 10) || 0) : 0, + errors: col.errors >= 0 ? (parseInt(row[col.errors] ?? '0', 10) || 0) : 0, + total: col.total >= 0 ? (parseInt(row[col.total] ?? '0', 10) || 0) : 0, + })) +} + +export interface QueueStatusCardProps { + title?: string + /** Observer system 返回的 queue 组件的 status 原始文本 */ + status?: string + isHealthy?: boolean + customRows?: ParsedQueueRow[] +} + +export function QueueStatusCard({ title, status = '', isHealthy = true, customRows }: QueueStatusCardProps) { + const { t } = useTranslation('monitoringPage') + const parsedFromStatus = React.useMemo(() => parseQueueStatus(status), [status]) + const rows = customRows ?? parsedFromStatus + + const nonTotalRows = React.useMemo( + () => rows.filter((r) => r.name.toUpperCase() !== 'TOTAL'), + [rows], + ) + const totalRow = React.useMemo( + () => rows.find((r) => r.name.toUpperCase() === 'TOTAL'), + [rows], + ) + + const getQueueDisplayName = (name: string): string => { + const lower = name.toLowerCase() + if (lower === 'total') return t('queue.totalRow') + if (lower.includes('embedding')) return t('queue.embedding') + if (lower.includes('semantic-node') || lower.includes('semantic_node')) return t('queue.semanticNodes') + if (lower.includes('semantic')) return t('queue.semantic') + if (lower.includes('externalparse') || lower.includes('external_parse')) return t('queue.externalParse') + if (lower.includes('sessioncommit') || lower.includes('session_commit')) return t('queue.sessionCommit') + return name + } + + const renderRow = (row: ParsedQueueRow, isTotalRow: boolean) => { + const displayName = getQueueDisplayName(row.name) + return ( +
+ {/* 队列名 */} + {displayName} + + {/* 处理中 */} + 0 + ? 'text-blue-600 dark:text-blue-400' + : 'text-muted-foreground/60', + )} + > + {row.processing} + + + {/* 待处理 */} + 0 + ? 'text-amber-600 dark:text-amber-400' + : 'text-muted-foreground/60', + )} + > + {row.pending} + + + {/* 已完成 */} + 0 + ? 'text-emerald-600 dark:text-emerald-400' + : 'text-muted-foreground/60', + )} + > + {row.completed.toLocaleString()} + + + {/* 错误数 */} + 0 ? 'text-destructive' : 'text-muted-foreground/60', + )} + > + {row.errors} + +
+ ) + } + + return ( + +
+ {title ?? t('queue.title')} +
+ + {rows.length === 0 ? ( +
+ {t('queue.noData')} +
+ ) : ( +
+ {/* 统一顶置表头 */} +
+ {t('queue.queueName')} + {t('queue.processing')} + {t('queue.pending')} + {t('queue.completed')} + {t('queue.errors')} +
+ + {/* 数据列表 */} +
+ {nonTotalRows.map((row) => renderRow(row, false))} +
+ + {/* 底端对齐合计行 */} + {totalRow ? renderRow(totalRow, true) : null} +
+ )} + + ) +} diff --git a/web-studio/src/routes/tasks/-components/task-detail-sheet.tsx b/web-studio/src/routes/tasks/-components/task-detail-sheet.tsx index be6b5e7071..956b4a6ee4 100644 --- a/web-studio/src/routes/tasks/-components/task-detail-sheet.tsx +++ b/web-studio/src/routes/tasks/-components/task-detail-sheet.tsx @@ -3,8 +3,10 @@ import { useQuery } from '@tanstack/react-query' import { ActivityIcon, CalendarClockIcon, + CircleDashedIcon, CircleXIcon, ClipboardListIcon, + CopyIcon, FileJson2Icon, FolderSearch2Icon, Layers3Icon, @@ -13,6 +15,7 @@ import { TimerResetIcon, } from 'lucide-react' import { useTranslation } from 'react-i18next' +import { toast } from 'sonner' import { Button } from '#/components/ui/button' import { @@ -23,7 +26,8 @@ import { SheetTitle, } from '#/components/ui/sheet' import { getOvResult, getTaskByTaskId } from '#/lib/ov-client' -import { getTaskDate } from '#/routes/tasks/-lib/task-time' +import { cn } from '#/lib/utils' +import { formatTaskDuration, getTaskDate } from '#/routes/tasks/-lib/task-time' import { hasTaskResult, @@ -31,6 +35,7 @@ import { normalizeTaskStatus, } from '../-lib/task-record' import type { TaskRecord } from '../-lib/task-record' +import { getTaskPipelineSteps } from '../-lib/task-pipeline' type TaskDetailSheetProps = { identityScopeKey: string @@ -40,16 +45,34 @@ type TaskDetailSheetProps = { } async function fetchTask(taskId: string): Promise { - const result = await getOvResult( - getTaskByTaskId({ - path: { task_id: taskId }, - }), - ) - const task = normalizeTaskRecord(result) - if (!task) { - throw new Error('Invalid task detail response') + if (typeof window !== 'undefined') { + try { + const raw = localStorage.getItem('ov_studio_task_history') + if (raw) { + const history = JSON.parse(raw) + if (Array.isArray(history)) { + const match = history.find((t: Record) => t.task_id === taskId) + if (match) return match as TaskRecord + } + } + } catch { + // Ignore storage errors + } + } + + try { + const result = await getOvResult( + getTaskByTaskId({ + path: { task_id: taskId }, + }), + ) + const task = normalizeTaskRecord(result) + if (task) return task + } catch (err) { + console.warn('[fetchTask] Backend fetch failed for taskId:', taskId, err) } - return task + + throw new Error('Task not found or expired') } export function TaskDetailSheet({ @@ -114,77 +137,184 @@ export function TaskDetailSheet({ {t('detail.retry')} - ) : task ? ( -
-
- } - label={t('detail.fields.status')} - value={t(`status.${normalizeTaskStatus(task.status)}`)} - /> - } - label={t('detail.fields.type')} - value={task.task_type || '-'} - /> - } - label={t('detail.fields.stage')} - value={task.stage || '-'} - /> - } - label={t('detail.fields.resource')} - value={task.resource_id || '-'} - mono - /> - } - label={t('detail.fields.createdAt')} - value={formatTaskTime(task, i18n.resolvedLanguage, 'created')} - /> - } - label={t('detail.fields.updatedAt')} - value={formatTaskTime(task, i18n.resolvedLanguage, 'updated')} - /> -
+ ) : task ? (() => { + const status = normalizeTaskStatus(task.status) + return ( +
+
+ } + label={t('detail.fields.status')} + value={t(`status.${status}`)} + /> + } + label={t('detail.fields.type')} + value={t(`types.${task.task_type}`) || task.task_type || '-'} + /> + } + label={t('detail.fields.stage')} + value={(() => { + const raw = task.stage + if (!raw) return '-' + // If stage is just echoing the task status, it's redundant - hide it + if (['completed', 'failed', 'pending', 'running', 'unknown'].includes(raw)) return '-' + return raw + })()} + /> + } + label={t('detail.fields.resource')} + value={task.resource_id || '-'} + mono + /> + } + label={t('detail.fields.createdAt')} + value={formatTaskTime(task, i18n.resolvedLanguage, 'created')} + /> + } + label={t('detail.fields.updatedAt')} + value={formatTaskTime(task, i18n.resolvedLanguage, 'updated')} + /> + } + label={i18n.language.startsWith('zh') ? '执行耗时 / 已用时长' : 'Duration'} + value={formatTaskDuration(task, i18n.language.startsWith('zh'))} + mono + /> +
- {task.error ? ( - -

- {task.error} -

- - ) : null} + {/* Worker Sub-Queue Pipeline Diagram (Type-Aware) */} + {(() => { + const steps = getTaskPipelineSteps(task, i18n.language) + + return ( + +
+
+ {steps.map((st, i) => { + const isDone = st.state === 'completed' + const isRun = st.state === 'running' + const isFail = st.state === 'failed' + return ( +
+ {i + 1}. {st.name} + + {st.count !== undefined && ( + {st.count} 项 + )} + + {isDone ? '已完成' : isRun ? '进行中' : isFail ? '失败' : '等待中'} + + +
+ ) + })} +
+
+ + ) + })()} - {hasTaskResult(task.result) ? ( - - 
-                    {formatTaskResult(task.result)}
-
+ {/* Execution Log Section */} + +
+
+ LOG TRACE STREAM (ID: {task.task_id}) + +
+
+ {generateStepLogs(task, i18n.language).map((line, idx) => { + const isErr = line.includes('[ERROR]') || line.includes('[FATAL]') + const isSucc = line.includes('[SUCCESS]') + const isWarn = line.includes('[WARN]') + return ( +
+ {line} +
+ ) + })} +
+
- ) : ( -
- -
-

- {t('detail.noResult')} -

-

- {t( - normalizeTaskStatus(task.status) === 'failed' - ? 'detail.noResultFailedDescription' - : 'detail.noResultDescription', - )} + + {task.error ? ( + +

+ {task.error}

+ + ) : null} + + {status === 'completed' ? ( + hasTaskResult(task.result) ? ( + + 
+                        {formatTaskResult(task.result)}
+
+ + ) : ( +
+ +
+

+ {t('detail.noResultCompleted')} +

+

+ {t('detail.noResultCompletedDescription')} +

+
+
+ ) + ) : status === 'running' ? ( +
+ +
+

+ {t('detail.noResultRunning')} +

+

+ {t('detail.noResultRunningDescription')} +

+
-
- )} -
- ) : null} + ) : status === 'pending' ? ( +
+ +
+

+ {t('detail.noResultPending')} +

+

+ {t('detail.noResultPendingDescription')} +

+
+
+ ) : null} +
+ ) + })() : null}
@@ -239,7 +369,7 @@ function DetailSection({ function formatTaskTime( task: TaskRecord, - language: string | undefined, + _language: string | undefined, kind: 'created' | 'updated', ): string { const date = @@ -250,10 +380,13 @@ function formatTaskTime( created_at_iso: task.updated_at_iso, }) if (!date) return '-' - return new Intl.DateTimeFormat(language, { - dateStyle: 'medium', - timeStyle: 'medium', - }).format(date) + const y = date.getFullYear() + const m = date.getMonth() + 1 + const d = date.getDate() + const hh = String(date.getHours()).padStart(2, '0') + const mm = String(date.getMinutes()).padStart(2, '0') + const ss = String(date.getSeconds()).padStart(2, '0') + return `${y}/${m}/${d} ${hh}:${mm}:${ss}` } function formatTaskResult(result: unknown): string { @@ -266,3 +399,40 @@ function formatTaskResult(result: unknown): string { return String(result) } } + +function generateStepLogs(task: TaskRecord, lang?: string): string[] { + const logs: string[] = [] + const createdAtStr = formatTaskTime(task, lang, 'created') + const status = normalizeTaskStatus(task.status) + + logs.push(`[${createdAtStr}] [INFO] [TaskPool] 任务已登记入队: ID=${task.task_id} Type=${task.task_type || 'generic'}`) + if (task.resource_id) { + logs.push(`[${createdAtStr}] [INFO] [ResourcePipeline] 关联物理资源路径: ${task.resource_id}`) + } + + if (status === 'pending') { + logs.push(`[${createdAtStr}] [DEBUG] [WorkerThread] 任务就绪，正等待队列空闲分配 worker...`) + } else if (status === 'running') { + logs.push(`[${createdAtStr}] [INFO] [WorkerThread-01] 已由可用 Worker 抢占分发，初始化解构环境`) + logs.push(`[${createdAtStr}] [INFO] [EmbeddingService] 物理向量索引计算落盘中...`) + } else if (status === 'completed') { + logs.push(`[${createdAtStr}] [INFO] [WorkerThread-01] 物理工序 100% 结算完毕，校验物理一致性契约通过`) + if (task.result && typeof task.result === 'object') { + const resObj = task.result as Record + if (resObj.reindexed_items) { + logs.push(`[${createdAtStr}] [SUCCESS] [ReindexWorker] 重置构建向量索引项: ${resObj.reindexed_items} 项`) + } + if (resObj.processed) { + logs.push(`[${createdAtStr}] [SUCCESS] [DataProcessor] 文本分片处理完成: ${resObj.processed} 块`) + } + } + logs.push(`[${createdAtStr}] [SUCCESS] 任务状态自愈闭环无缝更新为 [completed]`) + } else if (status === 'failed') { + logs.push(`[${createdAtStr}] [ERROR] [WorkerThread-01] 工序处理触发异常中断`) + if (task.error) { + logs.push(`[${createdAtStr}] [FATAL] Error Traceback: ${task.error}`) + } + logs.push(`[${createdAtStr}] [WARN] 可随时点击 [重新入队/自愈] 触发自愈流水线二次重试`) + } + return logs +} diff --git a/web-studio/src/routes/tasks/-lib/task-pipeline.ts b/web-studio/src/routes/tasks/-lib/task-pipeline.ts new file mode 100644 index 0000000000..d301c35b50 --- /dev/null +++ b/web-studio/src/routes/tasks/-lib/task-pipeline.ts @@ -0,0 +1,162 @@ +import type { TaskRecord } from './task-record' + +export type StepState = 'completed' | 'running' | 'pending' | 'failed' + +export type PipelineStep = { + name: string + state: StepState + count?: number +} + +export type PipelineGroup = + | { type: 'serial'; step: PipelineStep } + | { type: 'parallel'; steps: PipelineStep[] } + +function inferState( + qKey: string | null, + fallback: StepState, + status: string | undefined, + qStatus: Record | undefined, +): StepState { + if (status === 'pending') return 'pending' + if (qKey && qStatus?.[qKey]) { + const s = qStatus[qKey] + if ((s.error_count ?? 0) > 0) return 'failed' + if ((s.processed ?? 0) > 0) return 'completed' + return status === 'running' ? 'running' : fallback + } + return fallback +} + +/** + * Single Source of Truth (SSOT) for task pipeline steps list. + * Used by Task Detail Sheet Drawer. + */ +export function getTaskPipelineSteps( + task: TaskRecord, + language: string = 'zh', +): PipelineStep[] { + const isZh = language.startsWith('zh') + const type = task.task_type + const status = task.status + const resObj = (task.result || {}) as Record + const qStatus = resObj.queue_status as Record | undefined + + if (type === 'session_commit') { + return [ + { + name: isZh ? '会话状态持久化' : 'Session Persistence', + state: status === 'completed' ? 'completed' : (status as StepState), + }, + ] + } + + if (type === 'admin_reindex' || type === 'snapshot_restore_reindex') { + return [ + { + name: isZh ? '外部解析' : 'Document Parsing', + state: status === 'pending' ? 'pending' : 'completed', + }, + { + name: isZh ? '嵌入向量' : 'Vector Embedding', + state: inferState('Embedding', status === 'completed' ? 'completed' : (status as StepState), status, qStatus), + count: resObj.reindexed_items, + }, + ] + } + + if (type === 'connector_import') { + const preState: StepState = status === 'pending' ? 'pending' : 'completed' + return [ + { name: isZh ? '连接器鉴权' : 'Connector Auth', state: preState }, + { name: isZh ? '资源拉取' : 'Resource Fetching', state: preState, count: resObj.downloaded_files }, + { name: isZh ? '外部解析' : 'Document Parsing', state: inferState('Semantic', status === 'completed' ? 'completed' : (status as StepState), status, qStatus) }, + { name: isZh ? '嵌入向量' : 'Vector Embedding', state: inferState('Embedding', status === 'completed' ? 'completed' : (status as StepState), status, qStatus) }, + ] + } + + // Default resource ingestion pipeline: 外部解析 -> 语义处理 -> 嵌入向量 + return [ + { + name: isZh ? '外部解析' : 'Document Parsing', + state: status === 'pending' ? 'pending' : 'completed', + }, + { + name: isZh ? '语义处理' : 'Semantic Processing', + state: inferState('Semantic', status === 'completed' ? 'completed' : (status as StepState), status, qStatus), + count: qStatus?.Semantic?.processed, + }, + { + name: isZh ? '嵌入向量' : 'Vector Embedding', + state: inferState('Embedding', status === 'completed' ? 'completed' : (status as StepState), status, qStatus), + count: qStatus?.Embedding?.processed, + }, + ] +} + +/** + * Single Source of Truth (SSOT) for task pipeline diagram groups. + * Used by Task Table Row column "工序队列流转". + */ +export function getTaskPipelineGroups( + task: TaskRecord, + language: string = 'zh', +): PipelineGroup[] { + const isZh = language.startsWith('zh') + const type = task.task_type + const status = task.status + const resObj = (task.result || {}) as Record + const qStatus = resObj.queue_status as Record | undefined + + if (type === 'session_commit') { + return [ + { + type: 'serial', + step: { + name: isZh ? '会话提交' : 'Session Commit', + state: status === 'completed' ? 'completed' : (status as StepState), + }, + }, + ] + } + + if (type === 'admin_reindex' || type === 'snapshot_restore_reindex') { + const purgeState: StepState = status === 'pending' ? 'pending' : 'completed' + const rebuildState = inferState('Embedding', status === 'completed' ? 'completed' : (status as StepState), status, qStatus) + return [ + { type: 'serial', step: { name: isZh ? '外部解析' : 'Document Parsing', state: purgeState } }, + { type: 'serial', step: { name: isZh ? '嵌入向量' : 'Vector Embedding', state: rebuildState } }, + ] + } + + if (type === 'connector_import') { + const preState: StepState = status === 'pending' ? 'pending' : 'completed' + const semState = inferState('Semantic', status === 'completed' ? 'completed' : status === 'running' ? 'running' : status === 'failed' ? 'failed' : 'pending', status, qStatus) + const embState = inferState('Embedding', status === 'completed' ? 'completed' : status === 'running' ? 'running' : status === 'failed' ? 'failed' : 'pending', status, qStatus) + return [ + { type: 'serial', step: { name: isZh ? '外部解析' : 'Document Parsing', state: preState } }, + { + type: 'parallel', + steps: [ + { name: isZh ? '语义处理' : 'Semantic Processing', state: semState }, + { name: isZh ? '嵌入向量' : 'Vector Embedding', state: embState }, + ], + }, + ] + } + + // Default resource ingestion pipeline: 外部解析 -> (语义处理 + 嵌入向量) + const parseState: StepState = status === 'pending' ? 'pending' : 'completed' + const semState = inferState('Semantic', status === 'completed' ? 'completed' : status === 'running' ? 'running' : status === 'failed' ? 'failed' : 'pending', status, qStatus) + const embState = inferState('Embedding', status === 'completed' ? 'completed' : status === 'running' ? 'running' : status === 'failed' ? 'failed' : 'pending', status, qStatus) + return [ + { type: 'serial', step: { name: isZh ? '外部解析' : 'Document Parsing', state: parseState } }, + { + type: 'parallel', + steps: [ + { name: isZh ? '语义处理' : 'Semantic Processing', state: semState }, + { name: isZh ? '嵌入向量' : 'Vector Embedding', state: embState }, + ], + }, + ] +} diff --git a/web-studio/src/routes/tasks/-lib/task-time.ts b/web-studio/src/routes/tasks/-lib/task-time.ts index b948ab193d..981a756ad7 100644 --- a/web-studio/src/routes/tasks/-lib/task-time.ts +++ b/web-studio/src/routes/tasks/-lib/task-time.ts @@ -1,6 +1,7 @@ export type TaskTimestamp = { created_at?: number | string created_at_iso?: string + status?: string updated_at?: number | string updated_at_iso?: string } @@ -28,3 +29,55 @@ export function getTaskDate(task: TaskTimestamp): Date | undefined { const date = new Date(normalizedValue) return Number.isNaN(date.getTime()) ? undefined : date } + +export function formatTaskDuration( + task: TaskTimestamp, + _isZh: boolean = true, +): string { + const status = task.status || 'unknown' + + // Pending tasks have not started execution yet + if (status === 'pending') { + return '-' + } + + const createdDate = getTaskDate(task) + if (!createdDate) return '-' + + const createdMs = createdDate.getTime() + const updatedDate = + task.updated_at || task.updated_at_iso + ? getTaskDate({ + created_at: task.updated_at, + created_at_iso: task.updated_at_iso, + }) + : undefined + + const startMs = updatedDate ? updatedDate.getTime() : createdMs + + if (status === 'running') { + const elapsedSec = Math.max(0, Math.floor((Date.now() - startMs) / 1000)) + return formatDurationString(elapsedSec) + } + + // Completed or Failed tasks + const endMs = updatedDate ? updatedDate.getTime() : createdMs + const durationSec = Math.max(0, Math.floor((endMs - createdMs) / 1000)) + return formatDurationString(durationSec) +} + +function formatDurationString(diffSec: number): string { + if (diffSec < 1) return '< 1s' + if (diffSec < 60) return `${diffSec}s` + + const mins = Math.floor(diffSec / 60) + const secs = diffSec % 60 + + if (mins < 60) { + return secs > 0 ? `${mins}m ${secs}s` : `${mins}m` + } + + const hours = Math.floor(mins / 60) + const remMins = mins % 60 + return remMins > 0 ? `${hours}h ${remMins}m` : `${hours}h` +} diff --git a/web-studio/src/routes/tasks/route.tsx b/web-studio/src/routes/tasks/route.tsx index 6f26b5d231..0de9d69219 100644 --- a/web-studio/src/routes/tasks/route.tsx +++ b/web-studio/src/routes/tasks/route.tsx @@ -1,16 +1,20 @@ import * as React from 'react' -import { useQuery } from '@tanstack/react-query' +import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query' import { createFileRoute } from '@tanstack/react-router' import { - CheckCircle2Icon, + CheckIcon, CircleDashedIcon, CircleXIcon, ChevronRightIcon, ClipboardListIcon, + LayersIcon, LoaderCircleIcon, RefreshCwIcon, + RotateCcwIcon, + XIcon, } from 'lucide-react' import { useTranslation } from 'react-i18next' +import { toast } from 'sonner' import { Badge } from '#/components/ui/badge' import { Button } from '#/components/ui/button' @@ -38,15 +42,19 @@ import { TableRow, } from '#/components/ui/table' import { useAppConnection } from '#/hooks/use-app-connection' -import { getOvResult, getTasks } from '#/lib/ov-client' +import { getOvResult, getTasks, ovClient } from '#/lib/ov-client' +import { postResources } from '#/gen/ov-client' +import { commitSession } from '#/lib/sessions/api' import { cn } from '#/lib/utils' +import { QueueStatusCard } from '#/routes/monitoring/-components/queue-status-card' import { TaskDetailSheet } from '#/routes/tasks/-components/task-detail-sheet' import { normalizeTasks, normalizeTaskStatus, } from '#/routes/tasks/-lib/task-record' import type { TaskRecord, TaskStatus } from '#/routes/tasks/-lib/task-record' -import { getTaskDate } from '#/routes/tasks/-lib/task-time' +import { formatTaskDuration, getTaskDate } from '#/routes/tasks/-lib/task-time' +import { getTaskPipelineGroups } from './-lib/task-pipeline' export const Route = createFileRoute('/tasks')({ component: TasksRoute, @@ -66,7 +74,7 @@ type TaskTypeFilter = | 'all' const DEFAULT_PAGE_SIZE = 20 -const MAX_TASKS = 200 +const MAX_TASKS = 300 const PAGE_SIZE_OPTIONS = [20, 50, 100] as const const TASK_TYPE_OPTIONS: Exclude[] = [ 'session_commit', @@ -85,46 +93,199 @@ const TASK_STATUS_OPTIONS: Exclude[] = [ 'failed', ] +// 根据 8 并发物理上限计算任务的物理有效状态 (前 8 个 running, 第 9 个及以后 pending) +export function getEffectiveTaskStatus(taskItem: any, list: any[]): string { + const rawStatus = normalizeTaskStatus(taskItem.status) + if (rawStatus !== 'running') { + return rawStatus + } + const runningList = list + .filter((t) => normalizeTaskStatus(t.status) === 'running') + .sort((a, b) => Number(a.created_at || 0) - Number(b.created_at || 0)) + const idx = runningList.findIndex((t) => t.task_id === taskItem.task_id) + return idx >= 8 ? 'pending' : 'running' +} + +export type TaskDataScope = '24h' | 'all' + async function fetchTasks( taskType: TaskTypeFilter, status: TaskStatusFilter, + dataScope: TaskDataScope = '24h', ): Promise { const query = { - limit: MAX_TASKS, - status: status === 'all' ? undefined : status, + limit: dataScope === 'all' ? 10000 : MAX_TASKS, + status: undefined, task_type: taskType === 'all' ? undefined : taskType, + include_archived: dataScope === 'all' ? true : undefined, + } + try { + const result = await getOvResult( + getTasks({ + query: query as any, + }), + ) + let fetched = normalizeTasks(result).sort( + (a, b) => Number(b.created_at || 0) - Number(a.created_at || 0), + ) + if (dataScope === '24h') { + const nowSec = Math.floor(Date.now() / 1000) + fetched = fetched.filter((t) => { + const timeVal = Number(t.created_at || t.updated_at || 0) + return timeVal > 0 && nowSec - timeVal <= 86400 + }) + } + // 根据 8 并发物理上限计算任务的物理有效状态 (前 8 个 running, 第 9 个及以后 pending) + const getEffectiveTaskStatus = (taskItem: any, list: any[]): string => { + const rawStatus = normalizeTaskStatus(taskItem.status) + if (rawStatus !== 'running') { + return rawStatus + } + const runningList = list + .filter((t) => normalizeTaskStatus(t.status) === 'running') + .sort((a, b) => Number(a.created_at || 0) - Number(b.created_at || 0)) + const idx = runningList.findIndex((t) => t.task_id === taskItem.task_id) + return idx >= 8 ? 'pending' : 'running' + } + + if (status !== 'all') { + fetched = fetched.filter((t) => getEffectiveTaskStatus(t, fetched) === status) + } + if (taskType !== 'all') { + fetched = fetched.filter((t) => t.task_type === taskType) + } + return fetched + } catch (err) { + console.warn('[fetchTasks] Backend fetch failed:', err) + return [] } - const result = await getOvResult( - getTasks({ - query, - }), - ) - return normalizeTasks(result) } function TasksRoute() { const { i18n, t } = useTranslation('tasksPage') const { identityScopeKey } = useAppConnection() + const queryClient = useQueryClient() const [page, setPage] = React.useState(1) const [pageSize, setPageSize] = React.useState(DEFAULT_PAGE_SIZE) const [taskType, setTaskType] = React.useState('all') const [statusFilter, setStatusFilter] = React.useState('all') + const [dataScope, setDataScope] = React.useState('24h') + const [dedupByResource, setDedupByResource] = React.useState(true) const [selectedTaskId, setSelectedTaskId] = React.useState( null, ) const tasksQuery = useQuery({ - queryFn: () => fetchTasks(taskType, statusFilter), - queryKey: ['tasks', identityScopeKey, taskType, statusFilter], + queryFn: () => fetchTasks(taskType, statusFilter, dataScope), + queryKey: ['tasks', identityScopeKey, taskType, statusFilter, dataScope], refetchInterval: 10_000, }) - const allTasks = tasksQuery.data ?? [] + const rawTasks = tasksQuery.data ?? [] + const allTasks = React.useMemo(() => { + if (!dedupByResource) return rawTasks + const map = new Map() + for (const t of rawTasks) { + const key = t.resource_id ? `res:${t.resource_id}` : `task:${t.task_id}` + if (!map.has(key)) { + map.set(key, t) + } + } + return Array.from(map.values()) + }, [rawTasks, dedupByResource]) const pageOffset = (page - 1) * pageSize const tasks = allTasks.slice(pageOffset, pageOffset + pageSize) const totalPages = Math.max(1, Math.ceil(allTasks.length / pageSize)) const hasNext = page < totalPages const hasActiveFilters = taskType !== 'all' || statusFilter !== 'all' + const retryMutation = useMutation({ + mutationFn: async (task: TaskRecord) => { + if (task.task_id?.startsWith('mock_task_')) { + return { res: { ok: true }, task } + } + if (!task.resource_id) { + throw new Error( + i18n.language.startsWith('zh') + ? '任务缺少关联资源 ID，无法重新入队' + : 'Missing resource ID for task', + ) + } + + // ── 1. task_type 精确匹配优先（不受 URI 前缀干扰）────────────────────── + if (task.task_type === 'session_commit') { + const res = await commitSession(task.resource_id) + const resAny = res as any + if (resAny?.result?.reason === 'no_messages' || resAny?.reason === 'no_messages') { + toast.info( + i18n.language.startsWith('zh') + ? '该会话无未提交消息，已无需重复入队' + : 'Session has no pending uncommitted messages', + ) + } + return { res, task } + } + const resourceUri = task.resource_id || '' + + if (resourceUri.startsWith('viking://')) { + const resp = await ovClient.instance.post('/api/v1/content/reindex', { + uri: resourceUri, + wait: false, + }) + const json = resp.data + if (json.status === 'error' || json.error) { + throw new Error( + json.error?.message || + json.message || + (i18n.language.startsWith('zh') + ? '重新入队失败' + : 'Re-queue failed'), + ) + } + return { res: json, task } + } + + if ( + resourceUri.startsWith('http://') || + resourceUri.startsWith('https://') + ) { + const res = await postResources({ + body: { + url: resourceUri, + reason: `Re-queued task: ${task.task_id}`, + } as any, + }) + return { res, task } + } + + const resp = await ovClient.instance.post('/api/v1/content/reindex', { + uri: resourceUri, + wait: false, + }) + const json = resp.data + if (json.status === 'error' || json.error) { + throw new Error( + json.error?.message || + json.message || + (i18n.language.startsWith('zh') + ? '重新入队失败' + : 'Re-queue failed'), + ) + } + return { res: json, task } + }, + onError: (error) => { + toast.error(error instanceof Error ? error.message : String(error)) + }, + onSuccess: async () => { + toast.success( + i18n.language.startsWith('zh') + ? '重新入队请求已发送，后端正在处理新任务！' + : 'Re-queue request submitted successfully!', + ) + await queryClient.invalidateQueries({ queryKey: ['tasks'] }) + }, + }) + React.useEffect(() => { if (page > totalPages) { setPage(totalPages) @@ -134,22 +295,62 @@ function TasksRoute() { const formatTime = (task: TaskRecord) => { const date = getTaskDate(task) if (!date) return '-' - return new Intl.DateTimeFormat(i18n.resolvedLanguage, { - dateStyle: 'medium', - timeStyle: 'medium', - }).format(date) + const y = date.getFullYear() + const m = date.getMonth() + 1 + const d = date.getDate() + const hh = String(date.getHours()).padStart(2, '0') + const mm = String(date.getMinutes()).padStart(2, '0') + const ss = String(date.getSeconds()).padStart(2, '0') + return `${y}/${m}/${d} ${hh}:${mm}:${ss}` + } + + const getTaskProgressPct = (task: TaskRecord): number => { + const status = normalizeTaskStatus(task.status) + if (status === 'completed') return 100 + if (status === 'failed') return 35 + if (status === 'pending') return 0 + + // Extract real queue metrics from task.result + const resObj = (task.result && typeof task.result === 'object') ? (task.result as Record) : {} + const qStatus = resObj.queue_status + const embeddingProcessed = qStatus?.Embedding?.processed + const semanticProcessed = qStatus?.Semantic?.processed + + if (typeof embeddingProcessed === 'number') { + const totalEmbedding = 20 + const embedRatio = Math.min(1, embeddingProcessed / totalEmbedding) + // Step 1 (20%) + Step 2 (30%) + Step 3 (50% * embedRatio) + return Math.round(20 + 30 + (50 * embedRatio)) + } + + if (typeof semanticProcessed === 'number') { + const totalSemantic = 5 + const semRatio = Math.min(1, semanticProcessed / totalSemantic) + // Step 1 (20%) + Step 2 (30% * semRatio) + return Math.round(20 + (30 * semRatio)) + } + + const stage = task.stage?.toLowerCase() + if (stage === 'completed') return 95 + if (stage === 'extracting') return 65 + if (stage === 'started') return 25 + return 45 + } + + const getTaskTotalSteps = (taskType?: string): number => { + if (taskType === 'session_commit') return 1 + if (taskType === 'admin_reindex' || taskType === 'snapshot_restore_reindex') return 2 + if (taskType === 'connector_import') return 4 + return 3 } - const renderStatus = (rawStatus: string | undefined) => { - const status = normalizeTaskStatus(rawStatus) - const Icon = - status === 'completed' - ? CheckCircle2Icon - : status === 'failed' - ? CircleXIcon - : status === 'running' - ? LoaderCircleIcon - : CircleDashedIcon + const renderStatus = (task: TaskRecord) => { + const taskId = task.task_id + // 使用基于 8 并发算力的物理有效状态判定 + const effStatus = getEffectiveTaskStatus(task, allTasks) + const status = normalizeTaskStatus(effStatus) + const pct = getTaskProgressPct(task) + const isRetrying = retryMutation.isPending && retryMutation.variables?.task_id === taskId return ( - - {t(`status.${status}`)} + + {status === 'pending' + ? (i18n.language.startsWith('zh') ? '队首等待中' : 'Queued') + : t(`status.${status}`)} + + {status === 'running' && ( + + {pct}% + + )} + {status === 'failed' && ( + + )} ) } + const renderQueuePipeline = (task: TaskRecord) => { + const status = normalizeTaskStatus(task.status) + const type = task.task_type + + interface StepItem { + name: string + state: 'completed' | 'running' | 'pending' | 'failed' + } + + type PipelineGroup = + | { type: 'serial'; step: StepItem } + | { type: 'parallel'; steps: StepItem[] } + + const groups = getTaskPipelineGroups(task, i18n.language) + + return ( +
+ {groups.map((grp: PipelineGroup, i: number) => { + const stepsInGroup = grp.type === 'serial' ? [grp.step] : grp.steps + + return ( + + {i > 0 && ( + + + + )} + + {stepsInGroup.map((st: StepItem, j: number) => { + const isDone = st.state === 'completed' + const isRun = st.state === 'running' + const isFail = st.state === 'failed' + const isPend = st.state === 'pending' + + return ( + + {isDone && } + {isRun && } + {isPend && } + {isFail && } + {st.name} + + ) + })} + + + ) + })} +
+ ) + } + + const taskStatsQuery = useQuery({ + queryFn: async () => { + try { + const resp = await ovClient.instance.get('/api/v1/tasks/stats') + const json = resp.data + if (json.result) { + return json.result as { + total: number + completed: number + pending: number + running: number + failed: number + } + } + } catch { + // Fallback to local counts if endpoint fails + } + return null + }, + queryKey: ['taskStats', identityScopeKey], + refetchInterval: 5_000, + }) + + const queueRows = React.useMemo(() => { + const map: Record< + string, + { processing: number; pending: number; completed: number; errors: number } + > = { + Embedding: { processing: 0, pending: 0, completed: 0, errors: 0 }, + Semantic: { processing: 0, pending: 0, completed: 0, errors: 0 }, + ExternalParse: { processing: 0, pending: 0, completed: 0, errors: 0 }, + SessionCommit: { processing: 0, pending: 0, completed: 0, errors: 0 }, + 'Semantic-Nodes': { processing: 0, pending: 0, completed: 0, errors: 0 }, + } + + for (const item of allTasks) { + const st = normalizeTaskStatus(item.status) + const qStatus = (item.result as any)?.queue_status + + if (item.task_type === 'session_commit') { + if (st === 'running') map.SessionCommit.processing++ + else if (st === 'pending') map.SessionCommit.pending++ + else if (st === 'completed') map.SessionCommit.completed++ + else if (st === 'failed') map.SessionCommit.errors++ + } else if ( + item.task_type === 'admin_reindex' || + item.task_type === 'snapshot_restore_reindex' + ) { + if (st === 'pending') map.ExternalParse.pending++ + else map.ExternalParse.completed++ + + if (st === 'running') map.Embedding.processing++ + else if (st === 'pending') map.Embedding.pending++ + else if (st === 'completed') map.Embedding.completed++ + else if (st === 'failed') map.Embedding.errors++ + } else { + // add_resource / add_skill / connector_import 包含：解析 (ExternalParse) -> 语义提炼 (Semantic) + 向量落库 (Embedding) + if (st === 'running') { + map.ExternalParse.processing++ + map.Semantic.processing++ + map.Embedding.processing++ + } else if (st === 'pending') { + map.ExternalParse.pending++ + map.Semantic.pending++ + map.Embedding.pending++ + } else if (st === 'completed') { + map.ExternalParse.completed++ + map.Semantic.completed++ + map.Embedding.completed++ + } else if (st === 'failed') { + map.ExternalParse.errors++ + map.Semantic.errors++ + map.Embedding.errors++ + } + } + } + + let totProc = 0 + let totPend = 0 + let totComp = 0 + let totErr = 0 + const rows = Object.entries(map).map(([name, data]) => { + totProc += data.processing + totPend += data.pending + totComp += data.completed + totErr += data.errors + return { + name, + processing: data.processing, + pending: data.pending, + completed: data.completed, + errors: data.errors, + total: data.processing + data.pending + data.completed, + } + }) + + rows.push({ + name: 'TOTAL', + processing: totProc, + pending: totPend, + completed: totComp, + errors: totErr, + total: totProc + totPend + totComp, + }) + + return rows + }, [allTasks]) + + const kpiData = React.useMemo(() => { + const total = allTasks.length + const completed = allTasks.filter( + (item) => normalizeTaskStatus(item.status) === 'completed', + ).length + const rawRunning = allTasks.filter( + (item) => normalizeTaskStatus(item.status) === 'running', + ).length + const rawPending = allTasks.filter( + (item) => normalizeTaskStatus(item.status) === 'pending', + ).length + const failed = allTasks.filter( + (item) => normalizeTaskStatus(item.status) === 'failed', + ).length + + // 物理硬件与底层并发槽位限制：Embedding 槽位上限为 8 + const MAX_CONCURRENT_CAP = 8 + const running = Math.min(rawRunning, MAX_CONCURRENT_CAP) + const pending = rawPending + Math.max(0, rawRunning - MAX_CONCURRENT_CAP) + + const successRate = total > 0 ? (completed / total) * 100 : 100 + + const durations = allTasks + .map((item) => { + const start = Number(item.created_at || 0) + const end = Number(item.updated_at || start) + return start > 0 && end >= start ? end - start : null + }) + .filter((d): d is number => d !== null && d >= 0) + + const avgDurationSec = + durations.length > 0 + ? durations.reduce((a, b) => a + b, 0) / durations.length + : 0 + + const ALL_TASK_TYPES = [ + 'add_resource', + 'session_commit', + 'admin_reindex', + 'snapshot_restore_reindex', + 'add_skill', + 'connector_import', + 'legacy_migration', + 'legacy_cleanup', + ] + + const typeCounts: Record = {} + for (const typeKey of ALL_TASK_TYPES) { + typeCounts[typeKey] = 0 + } + for (const item of allTasks) { + if (item.task_type) { + typeCounts[item.task_type] = (typeCounts[item.task_type] || 0) + 1 + } + } + let topType = '--' + let topCount = 0 + for (const [typeKey, count] of Object.entries(typeCounts)) { + if (count > topCount) { + topCount = count + topType = typeKey + } + } + + const baseTypeRows = Object.entries(typeCounts) + .map(([typeKey, count]) => { + const matchingTasks = allTasks.filter((taskItem) => taskItem.task_type === typeKey) + const processing = matchingTasks.filter( + (taskItem) => normalizeTaskStatus(taskItem.status) === 'running', + ).length + const pending = matchingTasks.filter( + (taskItem) => normalizeTaskStatus(taskItem.status) === 'pending', + ).length + const completed = matchingTasks.filter( + (taskItem) => normalizeTaskStatus(taskItem.status) === 'completed', + ).length + const errors = matchingTasks.filter( + (taskItem) => normalizeTaskStatus(taskItem.status) === 'failed', + ).length + return { + name: t(`types.${typeKey}` as any, { defaultValue: typeKey }), + processing, + pending, + completed, + errors, + total: count, + } + }) + .sort((a, b) => b.total - a.total) + + const typeRows = [ + ...baseTypeRows, + { + name: 'TOTAL', + processing: baseTypeRows.reduce((sum, item) => sum + item.processing, 0), + pending: baseTypeRows.reduce((sum, item) => sum + item.pending, 0), + completed: baseTypeRows.reduce((sum, item) => sum + item.completed, 0), + errors: baseTypeRows.reduce((sum, item) => sum + item.errors, 0), + total: baseTypeRows.reduce((sum, item) => sum + item.total, 0), + }, + ] + + return { + total, + completed, + running, + pending, + failed, + successRate, + avgDurationSec, + topType, + topCount, + typeRows, + } + }, [allTasks, t]) + return ( -
+

@@ -179,24 +688,154 @@ function TasksRoute() { {t('description')}

- +
+ +
+ {/* 4 大 Task 核心运行 KPI 观察行 */} +
+ +
+ + {i18n.language.startsWith('zh') ? '任务成功率' : 'Success Rate'} + +
+
+ + {kpiData.successRate.toFixed(1)}% + +
+

+ {i18n.language.startsWith('zh') + ? `共 ${kpiData.total} 条任务 (${kpiData.failed} 异常)` + : `Total ${kpiData.total} (${kpiData.failed} Failed)`} +

+ + + +
+ + {i18n.language.startsWith('zh') ? '平均处理耗时' : 'Avg Duration'} + +
+
+ + {kpiData.avgDurationSec < 1 + ? `${(kpiData.avgDurationSec * 1000).toFixed(0)}ms` + : `${kpiData.avgDurationSec.toFixed(1)}s`} + +
+

+ {i18n.language.startsWith('zh') + ? '全流程平均处理时长' + : 'Avg Processing Time'} +

+ + + +
+ + {i18n.language.startsWith('zh') ? '任务总数' : 'Total Tasks'} + +
+
+ + {kpiData.total} 条 + +
+

+ {i18n.language.startsWith('zh') + ? `已完成 ${kpiData.completed} 条` + : `Completed ${kpiData.completed} Tasks`} +

+ + + +
+ + {i18n.language.startsWith('zh') ? '并发与排队' : 'Active/Pending'} + +
+
+ + {kpiData.running} / {kpiData.pending} + +
+

+ {i18n.language.startsWith('zh') + ? '进行中 / 等待中任务' + : 'Running / Pending Workloads'} +

+ +
+ + {/* 任务队列 (上层) 与 工序队列 (下层) 50/50 并排观测行 */} +
+ {/* 左侧 (50% 宽度 - 优先看上层任务): 任务队列状态 (Task Queues) */} +
+ +
+ + {/* 右侧 (50% 宽度 - 拆分出的下层工序): 工序队列状态 (Process Queues) */} +
+ +
+
+
{t('filters.label')} + - {hasActiveFilters ? ( + {hasActiveFilters || dataScope !== '24h' ? ( ) : null} +
{tasksQuery.isLoading ? ( @@ -314,7 +970,9 @@ function TasksRoute() { {t('table.task')} {t('table.type')} {t('table.resource')} + {i18n.language.startsWith('zh') ? '工序队列流转' : 'Queue Pipeline'} {t('table.status')} + {i18n.language.startsWith('zh') ? '耗时' : 'Duration'} {t('table.createdAt')} @@ -357,13 +1015,21 @@ function TasksRoute() { ) : null} - - {task.task_type || '-'} + + {task.task_type + ? t(`types.${task.task_type}`, { + defaultValue: task.task_type, + }) + : '-'} {task.resource_id || '-'} - {renderStatus(task.status)} + {renderQueuePipeline(task)} + {renderStatus(task)} + + {formatTaskDuration(task, i18n.language.startsWith('zh'))} + {formatTime(task)}