diff --git a/api/v1/endpoints/analysis.py b/api/v1/endpoints/analysis.py index 6a31e6a5c..2c87e9f82 100644 --- a/api/v1/endpoints/analysis.py +++ b/api/v1/endpoints/analysis.py @@ -698,6 +698,8 @@ def _format_sse_event(event_type: str, data: Dict[str, Any]) -> str: def _load_history_run_flow_by_query_id( query_id: str, *, + code: Optional[str] = None, + report_type: Optional[str] = None, fail_open: bool = False, ) -> Optional[RunFlowSnapshot]: try: @@ -705,7 +707,11 @@ def _load_history_run_flow_by_query_id( from src.services.history_service import HistoryService service = HistoryService(DatabaseManager.get_instance()) - return service.resolve_and_get_run_flow(query_id) + return service.resolve_and_get_run_flow( + query_id, + code=code, + report_type=report_type, + ) except Exception as e: if fail_open: logger.debug( @@ -740,8 +746,16 @@ def get_task_run_flow(task_id: str) -> RunFlowSnapshot: if task: if task.status == TaskStatusEnum.COMPLETED: + task_report_type = _history_report_type_for_task_flow( + getattr(task, "report_type", None) + ) + task_stock_code = _safe_task_flow_text(getattr(task, "stock_code", None), max_length=32) + if task_report_type == "market_review": + task_stock_code = "MARKET" history_snapshot = _load_history_run_flow_by_query_id( task_id, + code=task_stock_code, + report_type=task_report_type, fail_open=True, ) if history_snapshot is not None: @@ -759,6 +773,31 @@ def get_task_run_flow(task_id: str) -> RunFlowSnapshot: raise api_error(404, "not_found", f"任务 {task_id} 不存在或已过期") +def _safe_task_flow_text(value: Any, *, max_length: int) -> Optional[str]: + if value is None: + return None + text = str(value).strip() + if not text: + return None + return text[:max_length] + + +def _history_report_type_for_task_flow(value: Any) -> Optional[str]: + text = _safe_task_flow_text(value, max_length=64) + if text is None: + return None + normalized = text.lower().strip().replace("-", "_") + aliases = { + "detailed": "full", + "simple": "simple", + "full": "full", + "brief": "brief", + "market": "market_review", + "market_review": "market_review", + } + return aliases.get(normalized, normalized) + + def _datetime_to_iso(value: Any) -> Optional[str]: if isinstance(value, datetime): return value.isoformat() diff --git a/api/v1/schemas/run_flow.py b/api/v1/schemas/run_flow.py index da9ecbf0b..060db7a16 100644 --- a/api/v1/schemas/run_flow.py +++ b/api/v1/schemas/run_flow.py @@ -57,7 +57,7 @@ class RunFlowNode(BaseModel): started_at: Optional[str] = Field(None, description="ISO timestamp when the node started") ended_at: Optional[str] = Field(None, description="ISO timestamp when the node ended") duration_ms: Optional[int] = Field(None, ge=0, description="Node duration in milliseconds") - attempts: Optional[int] = Field(None, ge=1, description="Attempt count represented by this node") + attempts: Optional[int] = Field(None, ge=0, description="Attempt count represented by this node") record_count: Optional[int] = Field(None, ge=0, description="Returned record count") message: Optional[str] = Field(None, description="Short sanitized status message") metadata: Dict[str, Any] = Field(default_factory=dict, description="Sanitized low-sensitivity metadata") diff --git a/apps/dsa-web/src/components/run-flow/RunFlowGraph.tsx b/apps/dsa-web/src/components/run-flow/RunFlowGraph.tsx index 90dc24b73..7e0886120 100644 --- a/apps/dsa-web/src/components/run-flow/RunFlowGraph.tsx +++ b/apps/dsa-web/src/components/run-flow/RunFlowGraph.tsx @@ -1,6 +1,7 @@ import type React from 'react'; import { useMemo, useId } from 'react'; -import { Badge, StatusDot, Tooltip } from '../common'; +import { ChevronDown, ChevronRight } from 'lucide-react'; +import { Badge, StatusDot } from '../common'; import { useUiLanguage } from '../../contexts/UiLanguageContext'; import type { RunFlowEdge, RunFlowLane, RunFlowNode, RunFlowStatus } from '../../types/runFlow'; import { @@ -13,12 +14,16 @@ import { RUN_FLOW_STATUS_STYLE, } from './utils'; +type RunFlowT = ReturnType['t']; + interface RunFlowGraphProps { lanes: RunFlowLane[]; nodes: RunFlowNode[]; edges: RunFlowEdge[]; selectedNodeId?: string | null; + expandedNodeIds?: Set; onSelectNode?: (node: RunFlowNode) => void; + onToggleExpanded?: (nodeId: string) => void; } type PositionedNode = RunFlowNode & { @@ -28,9 +33,17 @@ type PositionedNode = RunFlowNode & { height: number; row: number; laneIndex: number; + compact?: boolean; + expandedGroupId?: string; }; +interface DataSourceBlock { + id: string; + nodes: RunFlowNode[]; +} + type EdgePort = 'top' | 'right' | 'bottom' | 'left'; +type EdgeFocusLevel = 'none' | 'direct' | 'internal'; interface PortPoint { x: number; @@ -38,11 +51,30 @@ interface PortPoint { side: EdgePort; } -const LANE_WIDTH = 292; -const NODE_WIDTH = 244; -const NODE_HEIGHT = 124; +interface LaneMetrics { + laneWidth: number; + nodeWidth: number; +} + +const DEFAULT_LANE_WIDTH = 260; +const DEFAULT_NODE_WIDTH = 224; +const LANE_METRICS: Record = { + entry: { laneWidth: 220, nodeWidth: 188 }, + data_source: { laneWidth: 292, nodeWidth: 244 }, + analysis: { laneWidth: 260, nodeWidth: 224 }, + artifact: { laneWidth: 220, nodeWidth: 188 }, +}; +const NODE_HEIGHT = 112; +const COMPACT_NODE_HEIGHT = 96; const HEADER_HEIGHT = 42; -const ROW_HEIGHT = 144; +const ROW_HEIGHT = 140; +const ENTRY_ROW_HEIGHT = 152; +const ARTIFACT_ROW_HEIGHT = 152; +const DATA_SOURCE_ATTEMPT_GAP = 42; +const DATA_SOURCE_BLOCK_GAP = 40; +const DATA_SOURCE_GROUP_X_PADDING = 18; +const DATA_SOURCE_GROUP_TOP_PADDING = 18; +const DATA_SOURCE_GROUP_BOTTOM_PADDING = 18; const LEFT_PADDING = 20; const TOP_PADDING = 18; const BOTTOM_PADDING = 30; @@ -55,6 +87,30 @@ const getEdgeStroke = (status: RunFlowStatus): string => { return 'hsl(var(--muted-text))'; }; +const getEdgeFocusRank = (level: EdgeFocusLevel): number => { + if (level === 'internal') return 2; + if (level === 'direct') return 1; + return 0; +}; + +const getEdgeStrokeWidth = (edge: RunFlowEdge, focusLevel: EdgeFocusLevel): number => { + const isFallbackPath = edge.kind === 'fallback' || edge.kind === 'retry'; + if (focusLevel === 'internal') { + return isFallbackPath ? 3.5 : 3; + } + if (focusLevel === 'direct') { + return isFallbackPath ? 3 : 2.4; + } + return isFallbackPath ? 2.5 : 1.75; +}; + +const getEdgeOpacity = (selectedNodeId: string | null | undefined, focusLevel: EdgeFocusLevel): number => { + if (!selectedNodeId) return 0.68; + if (focusLevel === 'internal') return 0.95; + if (focusLevel === 'direct') return 0.82; + return 0.18; +}; + const findAvailableRow = (occupiedRows: Set, preferredRow: number): number => { const safePreferred = Math.max(0, preferredRow); for (let distance = 0; distance < 1000; distance += 1) { @@ -82,6 +138,43 @@ const getCenteredTrackOffset = (total: number, index: number, step = 12): number (index - (total - 1) / 2) * step ); +const getLaneMetrics = (laneId: string): LaneMetrics => ( + LANE_METRICS[laneId] || { laneWidth: DEFAULT_LANE_WIDTH, nodeWidth: DEFAULT_NODE_WIDTH } +); + +const getLaneRowHeight = (laneId: string): number => ( + laneId === 'entry' ? ENTRY_ROW_HEIGHT : (laneId === 'artifact' ? ARTIFACT_ROW_HEIGHT : ROW_HEIGHT) +); + +const isExpandableNode = (node: RunFlowNode): boolean => node.metadata?.topologyGroup === 'provider_attempts'; + +const getEdgeLabel = (label: string | null | undefined, t: RunFlowT): string | null => { + if (!label) return null; + if (label === '调用') return t('runFlow.edgeLabel.invoke'); + if (label === '详情') return t('runFlow.edgeLabel.details'); + return label; +}; + +const metadataString = (node: RunFlowNode, key: string): string | null => { + const value = node.metadata?.[key]; + return typeof value === 'string' && value.trim() ? value.trim() : null; +}; + +const dataTypeFromNode = (node: RunFlowNode): string | null => metadataString(node, 'data_type'); + +const topologyParentIdFromNode = (node: RunFlowNode): string | null => metadataString(node, 'topologyParentId'); + +const topologyRoleFromNode = (node: RunFlowNode): string | null => metadataString(node, 'topologyRole'); + +const topologyOrderFromNode = (node: RunFlowNode): number | null => { + const value = node.metadata?.topologyOrder; + return typeof value === 'number' && Number.isFinite(value) ? value : null; +}; + +const isExpandedProviderGroup = (node: RunFlowNode, expandedNodeIds?: Set): boolean => ( + isExpandableNode(node) && (expandedNodeIds?.has(node.id) || node.metadata?.expanded === true) +); + const portPoint = (node: PositionedNode, side: EdgePort, offset = 0): PortPoint => { if (side === 'top') { return { x: node.x + node.width / 2 + offset, y: node.y, side }; @@ -110,7 +203,7 @@ const chooseEdgePorts = ( : { startSide: 'top', endSide: 'bottom', vertical: true }; } - if ((edge.kind === 'fallback' || edge.kind === 'retry') && isVerticalRelation && Math.abs(to.x - from.x) < LANE_WIDTH * 1.25) { + if ((edge.kind === 'fallback' || edge.kind === 'retry') && isVerticalRelation && Math.abs(to.x - from.x) < DEFAULT_LANE_WIDTH * 1.25) { return to.y >= from.y ? { startSide: 'bottom', endSide: 'top', vertical: true } : { startSide: 'top', endSide: 'bottom', vertical: true }; @@ -162,12 +255,70 @@ const compareLaneNodes = ( return getNodeDisplayOrder(left, leftOriginal) - getNodeDisplayOrder(right, rightOriginal); }; +const buildDataSourceBlocks = ( + laneNodes: RunFlowNode[], + originalIndex: Map, + expandedNodeIds?: Set, +): DataSourceBlock[] => { + const expandedGroupByDataType = new Map(); + const expandedGroupById = new Map(); + laneNodes.forEach((node) => { + const dataType = dataTypeFromNode(node); + if (dataType && isExpandedProviderGroup(node, expandedNodeIds)) { + expandedGroupByDataType.set(dataType, node); + expandedGroupById.set(node.id, node); + } + }); + + const attemptGroupIdByNodeId = new Map(); + const attemptsByGroupId = new Map(); + laneNodes.forEach((node) => { + if (isExpandableNode(node)) { + return; + } + const explicitParentId = topologyParentIdFromNode(node); + const providerAttemptLike = topologyRoleFromNode(node) === 'provider_attempt' + || Boolean(explicitParentId) + || node.id.startsWith('provider_'); + const dataType = dataTypeFromNode(node); + const fallbackGroup = dataType ? expandedGroupByDataType.get(dataType) : null; + const groupId = explicitParentId && expandedGroupById.has(explicitParentId) + ? explicitParentId + : (providerAttemptLike ? fallbackGroup?.id : undefined); + if (!groupId || !expandedGroupById.has(groupId)) { + return; + } + const attempts = attemptsByGroupId.get(groupId) || []; + attempts.push(node); + attemptsByGroupId.set(groupId, attempts); + attemptGroupIdByNodeId.set(node.id, groupId); + }); + + const topLevelNodes = laneNodes + .filter((node) => !attemptGroupIdByNodeId.has(node.id)) + .sort((left, right) => compareLaneNodes('data_source', left, right, originalIndex)); + + return topLevelNodes.map((node) => { + if (!isExpandedProviderGroup(node, expandedNodeIds)) { + return { id: node.id, nodes: [node] }; + } + const attempts = [...(attemptsByGroupId.get(node.id) || [])].sort((left, right) => ( + (topologyOrderFromNode(left) ?? Number.MAX_SAFE_INTEGER) - (topologyOrderFromNode(right) ?? Number.MAX_SAFE_INTEGER) + || (nodeTimeOrder(left) ?? Number.MAX_SAFE_INTEGER) - (nodeTimeOrder(right) ?? Number.MAX_SAFE_INTEGER) + || getNodeDisplayOrder(left, originalIndex.get(left.id) ?? 0) - getNodeDisplayOrder(right, originalIndex.get(right.id) ?? 0) + )); + return { id: node.id, nodes: [node, ...attempts] }; + }); +}; + export const RunFlowGraph: React.FC = ({ lanes, nodes, edges, selectedNodeId, + expandedNodeIds, onSelectNode, + onToggleExpanded, }) => { const arrowId = useId().replace(/:/g, '-'); const { language, t } = useUiLanguage(); @@ -189,8 +340,12 @@ export const RunFlowGraph: React.FC = ({ const originalIndex = new Map(); const nodeById = new Map(); const laneIndexById = new Map(); + const laneOffsets = new Map(); + let nextLaneOffset = LEFT_PADDING; laneList.forEach((lane, index) => { laneIndexById.set(lane.id, index); + laneOffsets.set(lane.id, nextLaneOffset); + nextLaneOffset += getLaneMetrics(lane.id).laneWidth; }); nodes.forEach((node, index) => { const items = grouped.get(node.lane) || []; @@ -200,6 +355,18 @@ export const RunFlowGraph: React.FC = ({ nodeById.set(node.id, node); }); + const dataSourceBlocks = buildDataSourceBlocks(grouped.get('data_source') || [], originalIndex, expandedNodeIds); + const dataSourceNodeSequence = dataSourceBlocks.flatMap((block) => block.nodes); + const expandedGroupIdByAttemptId = new Map(); + dataSourceBlocks.forEach((block) => { + if (block.nodes.length <= 1 || !isExpandedProviderGroup(block.nodes[0], expandedNodeIds)) { + return; + } + block.nodes.slice(1).forEach((node) => { + expandedGroupIdByAttemptId.set(node.id, block.nodes[0].id); + }); + }); + const validEdges = edges.filter((edge) => nodeById.has(edge.from) && nodeById.has(edge.to)); const incomingByNode = new Map(); const outgoingByNode = new Map(); @@ -210,9 +377,11 @@ export const RunFlowGraph: React.FC = ({ const laneOrderByNode = new Map(); laneList.forEach((lane) => { - const laneNodes = [...(grouped.get(lane.id) || [])].sort((left, right) => ( - compareLaneNodes(lane.id, left, right, originalIndex) - )); + const laneNodes = lane.id === 'data_source' + ? dataSourceNodeSequence + : [...(grouped.get(lane.id) || [])].sort((left, right) => ( + compareLaneNodes(lane.id, left, right, originalIndex) + )); laneNodes.forEach((node, index) => { laneOrderByNode.set(node.id, index); }); @@ -253,29 +422,83 @@ export const RunFlowGraph: React.FC = ({ }; const positioned = new Map(); - let maxRow = 0; + let maxY = HEADER_HEIGHT + TOP_PADDING; laneList.forEach((lane, lanePosition) => { + const metrics = getLaneMetrics(lane.id); + if (lane.id === 'data_source') { + let yCursor = HEADER_HEIGHT + TOP_PADDING; + let row = 0; + dataSourceBlocks.forEach((block, blockIndex) => { + block.nodes.forEach((node, nodeIndex) => { + const compact = nodeIndex > 0 && expandedGroupIdByAttemptId.has(node.id); + const height = compact ? COMPACT_NODE_HEIGHT : NODE_HEIGHT; + positioned.set(node.id, { + ...node, + x: laneOffsets.get(lane.id) ?? LEFT_PADDING, + y: yCursor, + width: metrics.nodeWidth, + height, + row, + laneIndex: lanePosition, + compact, + expandedGroupId: expandedGroupIdByAttemptId.get(node.id), + }); + maxY = Math.max(maxY, yCursor + height); + yCursor += height; + yCursor += nodeIndex < block.nodes.length - 1 + ? DATA_SOURCE_ATTEMPT_GAP + : (blockIndex < dataSourceBlocks.length - 1 ? DATA_SOURCE_BLOCK_GAP : 0); + row += 1; + }); + }); + maxY = Math.max(maxY, yCursor); + return; + } + const laneNodes = [...(grouped.get(lane.id) || [])].sort((left, right) => ( resolvePreferredRow(left.id) - resolvePreferredRow(right.id) || compareLaneNodes(lane.id, left, right, originalIndex) )); const occupiedRows = new Set(); + const rowHeight = getLaneRowHeight(lane.id); laneNodes.forEach((node) => { const row = findAvailableRow(occupiedRows, resolvePreferredRow(node.id)); + const y = HEADER_HEIGHT + TOP_PADDING + row * rowHeight; occupiedRows.add(row); - maxRow = Math.max(maxRow, row); positioned.set(node.id, { ...node, - x: lanePosition * LANE_WIDTH + LEFT_PADDING, - y: HEADER_HEIGHT + TOP_PADDING + row * ROW_HEIGHT, - width: NODE_WIDTH, + x: laneOffsets.get(lane.id) ?? LEFT_PADDING, + y, + width: metrics.nodeWidth, height: NODE_HEIGHT, row, laneIndex: lanePosition, }); + maxY = Math.max(maxY, y + NODE_HEIGHT); }); }); + const expandedGroups = dataSourceBlocks + .map((block) => { + if (block.nodes.length <= 1 || !isExpandedProviderGroup(block.nodes[0], expandedNodeIds)) { + return null; + } + const groupNode = positioned.get(block.nodes[0].id); + const lastNode = positioned.get(block.nodes[block.nodes.length - 1].id); + if (!groupNode || !lastNode) return null; + return { + id: groupNode.id, + x: groupNode.x - DATA_SOURCE_GROUP_X_PADDING, + y: groupNode.y - DATA_SOURCE_GROUP_TOP_PADDING, + width: groupNode.width + DATA_SOURCE_GROUP_X_PADDING * 2, + height: lastNode.y + lastNode.height - groupNode.y + DATA_SOURCE_GROUP_TOP_PADDING + DATA_SOURCE_GROUP_BOTTOM_PADDING, + }; + }) + .filter((item): item is NonNullable => Boolean(item)); + const expandedGroupBottom = expandedGroups.reduce((bottom, group) => ( + Math.max(bottom, group.y + group.height) + ), 0); + const sortEdgesForAnchors = (edgeItems: RunFlowEdge[], fromNodeId: string) => [...edgeItems].sort((left, right) => { const leftTarget = nodeById.get(left.to); const rightTarget = nodeById.get(right.to); @@ -314,10 +537,32 @@ export const RunFlowGraph: React.FC = ({ positioned, incomingAnchors, outgoingAnchors, - width: Math.max(laneList.length * LANE_WIDTH + LEFT_PADDING, LANE_WIDTH), - height: HEADER_HEIGHT + TOP_PADDING + (maxRow + 1) * ROW_HEIGHT + BOTTOM_PADDING, + laneOffsets, + expandedGroups, + width: Math.max(nextLaneOffset + LEFT_PADDING, DEFAULT_LANE_WIDTH), + height: Math.max(maxY, expandedGroupBottom) + BOTTOM_PADDING, }; - }, [edges, laneList, nodes]); + }, [edges, expandedNodeIds, laneList, nodes]); + + const selectedNode = selectedNodeId ? layout.positioned.get(selectedNodeId) : null; + const selectedDataType = selectedNode ? dataTypeFromNode(selectedNode) : null; + const selectedProviderGroup = selectedDataType + ? Array.from(layout.positioned.values()).find((node) => ( + isExpandedProviderGroup(node, expandedNodeIds) && dataTypeFromNode(node) === selectedDataType + )) + : null; + const selectedRelatedNodeIds = new Set(); + if (selectedNodeId) { + selectedRelatedNodeIds.add(selectedNodeId); + } + if (selectedProviderGroup) { + selectedRelatedNodeIds.add(selectedProviderGroup.id); + Array.from(layout.positioned.values()).forEach((node) => { + if (dataTypeFromNode(node) === selectedDataType && (node.id === selectedProviderGroup.id || node.expandedGroupId === selectedProviderGroup.id)) { + selectedRelatedNodeIds.add(node.id); + } + }); + } const edgePaths = edges .map((edge, edgeIndex) => { @@ -337,17 +582,52 @@ export const RunFlowGraph: React.FC = ({ const start = portPoint(from, ports.startSide, startOffset); const end = portPoint(to, ports.endSide, endOffset); const path = orthogonalPath(start, end); - const relatedToSelected = Boolean(selectedNodeId && (edge.from === selectedNodeId || edge.to === selectedNodeId)); + const fromInSelectedGroup = selectedRelatedNodeIds.has(edge.from); + const toInSelectedGroup = selectedRelatedNodeIds.has(edge.to); + const internallyRelated = Boolean(selectedProviderGroup && fromInSelectedGroup && toInSelectedGroup); + const directlyRelated = Boolean( + edge.from === selectedNodeId + || edge.to === selectedNodeId + || (selectedProviderGroup && (fromInSelectedGroup || toInSelectedGroup)), + ); + const focusLevel: EdgeFocusLevel = selectedNodeId + ? (internallyRelated ? 'internal' : (directlyRelated ? 'direct' : 'none')) + : 'none'; return { edge, path, - labelX: (start.x + end.x) / 2, - labelY: (start.y + end.y) / 2 - 6, - relatedToSelected, + labelX: ports.vertical ? Math.max(start.x, end.x) + 10 : (start.x + end.x) / 2, + labelY: ports.vertical ? (start.y + end.y) / 2 + 4 : (start.y + end.y) / 2 - 8, + labelAnchor: ports.vertical ? ('start' as const) : ('middle' as const), + focusLevel, + relatedToSelected: focusLevel !== 'none', }; }) .filter((item): item is NonNullable => Boolean(item)); + const edgePathViews = edgePaths.reduce>((items, item) => { + const displayLabel = getEdgeLabel(item.edge.label, t); + const labelKey = `${item.edge.to}:${displayLabel || ''}`; + const duplicateLabel = items.some((existing) => ( + existing.relatedToSelected + && getEdgeLabel(existing.edge.label, t) + && `${existing.edge.to}:${getEdgeLabel(existing.edge.label, t)}` === labelKey + )); + items.push({ + ...item, + displayLabel, + showLabel: Boolean( + displayLabel + && (!selectedNodeId || item.relatedToSelected || item.edge.kind === 'fallback' || item.edge.kind === 'retry') + && !duplicateLabel, + ), + }); + return items; + }, []).sort((left, right) => getEdgeFocusRank(left.focusLevel) - getEdgeFocusRank(right.focusLevel)); + return (
@@ -379,71 +659,112 @@ export const RunFlowGraph: React.FC = ({ - + - {edgePaths.map(({ edge, path, labelX, labelY, relatedToSelected }) => ( + {edgePathViews.map(({ edge, path, labelX, labelY, labelAnchor, focusLevel, showLabel, displayLabel }) => ( - {edge.label && (!selectedNodeId || relatedToSelected || edge.kind === 'fallback' || edge.kind === 'retry') ? ( + {showLabel ? ( - {compactText(edge.label, 22)} + {compactText(displayLabel, 22)} ) : null} ))} - {laneList.map((lane, index) => ( + {laneList.map((lane) => { + const metrics = getLaneMetrics(lane.id); + const left = layout.laneOffsets.get(lane.id) ?? LEFT_PADDING; + return ( ) : null} + {(dataQuality || Object.keys(contextCounts).length > 0 || Object.keys(contextStatusCounts).length > 0) ? ( +
+

{t('runFlow.nodeDetails.contextQuality')}

+
+ {typeof dataQuality?.overallScore === 'number' ? ( +
+
{t('runFlow.nodeDetails.overallScore')}
+
{dataQuality.overallScore}
+
+ ) : null} + {dataQuality?.level ? ( +
+
{t('runFlow.nodeDetails.qualityLevel')}
+
{dataQuality.level}
+
+ ) : null} + {[ + ['available', 'success', t('runFlow.nodeDetails.count.available')], + ['missing', null, t('runFlow.nodeDetails.count.missing')], + ['partial', null, t('runFlow.nodeDetails.count.partial')], + ['degraded', null, t('runFlow.nodeDetails.count.degraded')], + ['fallback', null, t('runFlow.nodeDetails.count.fallback')], + ['skipped', null, t('runFlow.nodeDetails.count.skipped')], + ].map(([key, statusKey, label]) => { + const count = contextCounts[key || ''] ?? (statusKey ? contextStatusCounts[statusKey] : contextStatusCounts[key || '']); + if (typeof count !== 'number') return null; + return ( +
+
{label}
+
{count}
+
+ ); + })} +
+ + {Object.keys(blockScores).length > 0 ? ( +
+

{t('runFlow.nodeDetails.blockScores')}

+
+ {Object.entries(blockScores).map(([key, score]) => ( +
+
{key}
+
{score}
+
+ ))} +
+
+ ) : null} +
+ ) : null} + {metadata.length > 0 ? (

{t('runFlow.nodeDetails.metadata')}

diff --git a/apps/dsa-web/src/components/run-flow/RunFlowPanel.tsx b/apps/dsa-web/src/components/run-flow/RunFlowPanel.tsx index 3533477dd..f4aba6474 100644 --- a/apps/dsa-web/src/components/run-flow/RunFlowPanel.tsx +++ b/apps/dsa-web/src/components/run-flow/RunFlowPanel.tsx @@ -159,14 +159,16 @@ export const RunFlowPanel: React.FC = ({ source, title }) => className="border-dashed" /> ) : ( -
+
= ({ source, title }) => }} />
-
+
{ expect(screen.getByText('日线降级')).toBeInTheDocument(); expect(screen.getByText('任务取消')).toBeInTheDocument(); - fireEvent.click(screen.getByRole('button', { name: '降级/重试' })); + fireEvent.click(screen.getByRole('button', { name: '降级回退/重试' })); expect(screen.getByText('日线降级')).toBeInTheDocument(); expect(screen.queryByText('任务创建')).not.toBeInTheDocument(); diff --git a/apps/dsa-web/src/components/run-flow/__tests__/RunFlowGraph.test.tsx b/apps/dsa-web/src/components/run-flow/__tests__/RunFlowGraph.test.tsx index f3e1795ef..b918695c4 100644 --- a/apps/dsa-web/src/components/run-flow/__tests__/RunFlowGraph.test.tsx +++ b/apps/dsa-web/src/components/run-flow/__tests__/RunFlowGraph.test.tsx @@ -39,10 +39,30 @@ const edges: RunFlowEdge[] = [ }, ]; +const positionedStyleFor = (testId: string): CSSStyleDeclaration => { + return screen.getByTestId(`${testId}-wrapper`).style; +}; + +const nodeStyleFor = (testId: string): CSSStyleDeclaration => { + return screen.getByTestId(testId).style; +}; + +const layoutRowFor = (testId: string): number => ( + Number(screen.getByTestId(testId).dataset.layoutRow) +); + +const topFor = (testId: string): number => ( + parseFloat(positionedStyleFor(testId).top) +); + +const heightFor = (testId: string): number => ( + parseFloat(positionedStyleFor(testId).height) +); + describe('RunFlowGraph', () => { it('renders auto-layered lanes, edge legend labels, and clickable nodes', () => { const onSelectNode = vi.fn(); - render( + const { container } = render( { expect(screen.getByText('入口')).toBeInTheDocument(); expect(screen.getByText('数据来源')).toBeInTheDocument(); - expect(screen.getByText('降级')).toBeInTheDocument(); + expect(screen.getAllByText('降级回退').length).toBeGreaterThan(0); expect(screen.getByText('降级输入')).toBeInTheDocument(); expect(screen.getByTestId('run-flow-node-news')).toHaveTextContent('开始'); expect(screen.getByTestId('run-flow-node-news')).toHaveTextContent('2026'); - expect(screen.getByRole('button', { name: '新闻舆情 节点，状态 Fallback' })).toBeInTheDocument(); + expect(screen.getByRole('button', { name: '新闻舆情 节点，状态 降级回退' })).toBeInTheDocument(); + const marker = container.querySelector('marker'); + expect(marker).toHaveAttribute('markerWidth', '4'); + expect(marker).toHaveAttribute('markerHeight', '4'); + expect(marker).toHaveAttribute('refX', '3.5'); + expect(marker?.querySelector('path')).toHaveAttribute('d', 'M 0 0 L 4 2 L 0 4 z'); + fireEvent.mouseEnter(screen.getByTestId('run-flow-node-news')); + expect(screen.queryByRole('tooltip')).not.toBeInTheDocument(); - fireEvent.click(screen.getByRole('button', { name: '新闻舆情 节点，状态 Fallback' })); + fireEvent.click(screen.getByRole('button', { name: '新闻舆情 节点，状态 降级回退' })); expect(onSelectNode).toHaveBeenCalledWith(expect.objectContaining({ id: 'news' })); }); @@ -120,7 +147,9 @@ describe('RunFlowGraph', () => { const paths = Array.from(container.querySelectorAll('svg g path')); - expect(paths.map((path) => path.getAttribute('opacity'))).toEqual(['0.9', '0.22', '0.22']); + const opacities = paths.map((path) => path.getAttribute('opacity')); + expect(opacities.filter((opacity) => opacity === '0.82')).toHaveLength(1); + expect(opacities.filter((opacity) => opacity === '0.18')).toHaveLength(2); expect(screen.getByText('调度输入')).toBeInTheDocument(); expect(screen.queryByText('报告输出')).not.toBeInTheDocument(); expect(screen.getByText('降级输出')).toBeInTheDocument(); @@ -206,6 +235,36 @@ describe('RunFlowGraph', () => { expect(screen.getByTestId('run-flow-node-quote')).toHaveAttribute('data-layout-row'); }); + it('keeps entry lane nodes at the standard height with roomier vertical rhythm', () => { + render( + , + ); + + expect(heightFor('run-flow-node-request')).toBe(112); + expect(heightFor('run-flow-node-task_queue')).toBe(112); + expect(topFor('run-flow-node-task_queue')).toBe(topFor('run-flow-node-request') + 152); + expect(nodeStyleFor('run-flow-node-request').height).toBe('112px'); + }); + it('routes same-lane vertical edges from card bottom to the next card top', () => { const verticalNodes: RunFlowNode[] = [ { @@ -230,6 +289,7 @@ describe('RunFlowGraph', () => { to: 'quote', kind: 'control', status: 'success', + label: '详情', }, ]; const { container } = render( @@ -237,17 +297,18 @@ describe('RunFlowGraph', () => { lanes={lanes} nodes={verticalNodes} edges={verticalEdges} + selectedNodeId="daily" />, ); const pathData = container.querySelector('svg g path')?.getAttribute('d') || ''; const pathNumbers = pathData.match(/-?\d+(?:\.\d+)?/g)?.map(Number) || []; const [startX, startY, endY] = pathNumbers; - const dailyNode = screen.getByTestId('run-flow-node-daily'); - const quoteNode = screen.getByTestId('run-flow-node-quote'); - const dailyCenterX = parseFloat(dailyNode.style.left) + parseFloat(dailyNode.style.width) / 2; - const dailyBottom = parseFloat(dailyNode.style.top) + parseFloat(dailyNode.style.minHeight); - const quoteTop = parseFloat(quoteNode.style.top); + const dailyStyle = positionedStyleFor('run-flow-node-daily'); + const quoteStyle = positionedStyleFor('run-flow-node-quote'); + const dailyCenterX = parseFloat(dailyStyle.left) + parseFloat(dailyStyle.width) / 2; + const dailyBottom = parseFloat(dailyStyle.top) + parseFloat(dailyStyle.height); + const quoteTop = parseFloat(quoteStyle.top); expect(pathData).toContain('V'); expect(pathData).not.toContain('C'); @@ -255,6 +316,10 @@ describe('RunFlowGraph', () => { expect(startY).toBeLessThan(endY); expect(startY).toBe(dailyBottom); expect(endY).toBe(quoteTop); + const label = screen.getByText('详情'); + expect(label).toHaveAttribute('text-anchor', 'start'); + expect(parseFloat(label.getAttribute('x') || '0')).toBeGreaterThan(startX); + expect(parseFloat(label.getAttribute('y') || '0')).toBeGreaterThan((startY + endY) / 2); }); it('routes cross-lane flow edges through side ports with orthogonal segments', () => { @@ -281,6 +346,7 @@ describe('RunFlowGraph', () => { to: 'llm', kind: 'data', status: 'success', + label: '跨泳道', }, ]; const { container } = render( @@ -288,18 +354,19 @@ describe('RunFlowGraph', () => { lanes={lanes} nodes={crossLaneNodes} edges={crossLaneEdges} + selectedNodeId="request" />, ); const pathData = container.querySelector('svg g path')?.getAttribute('d') || ''; const pathNumbers = pathData.match(/-?\d+(?:\.\d+)?/g)?.map(Number) || []; const [startX, startY, , endY, endX] = pathNumbers; - const requestNode = screen.getByTestId('run-flow-node-request'); - const llmNode = screen.getByTestId('run-flow-node-llm'); - const requestRight = parseFloat(requestNode.style.left) + parseFloat(requestNode.style.width); - const requestCenterY = parseFloat(requestNode.style.top) + parseFloat(requestNode.style.minHeight) / 2; - const llmLeft = parseFloat(llmNode.style.left); - const llmCenterY = parseFloat(llmNode.style.top) + parseFloat(llmNode.style.minHeight) / 2; + const requestStyle = positionedStyleFor('run-flow-node-request'); + const llmStyle = positionedStyleFor('run-flow-node-llm'); + const requestRight = parseFloat(requestStyle.left) + parseFloat(requestStyle.width); + const requestCenterY = parseFloat(requestStyle.top) + parseFloat(requestStyle.height) / 2; + const llmLeft = parseFloat(llmStyle.left); + const llmCenterY = parseFloat(llmStyle.top) + parseFloat(llmStyle.height) / 2; expect(pathData).toContain('H'); expect(pathData).toContain('V'); @@ -308,6 +375,9 @@ describe('RunFlowGraph', () => { expect(startY).toBe(requestCenterY); expect(endX).toBe(llmLeft); expect(endY).toBe(llmCenterY); + const label = screen.getByText('跨泳道'); + expect(label).toHaveAttribute('text-anchor', 'middle'); + expect(parseFloat(label.getAttribute('y') || '0')).toBeLessThan((startY + endY) / 2); }); it('orders data-source lane cards by their observed timestamps', () => { @@ -353,4 +423,426 @@ describe('RunFlowGraph', () => { Number(screen.getByTestId('run-flow-node-late-news').dataset.layoutRow), ); }); + + it('keeps compact lane and card widths consistent across lanes', () => { + const laneWidthNodes: RunFlowNode[] = [ + { + id: 'request', + lane: 'entry', + kind: 'entry', + label: '用户请求', + status: 'success', + }, + { + id: 'news', + lane: 'data_source', + kind: 'data_source', + label: '新闻舆情', + status: 'success', + provider: 'TushareFetcher -> AkshareFetcher -> TushareFetcher -> AkshareFetcher', + }, + { + id: 'save', + lane: 'artifact', + kind: 'artifact', + label: '保存报告', + status: 'success', + }, + { + id: 'notification', + lane: 'artifact', + kind: 'notification', + label: '推送通知 · report', + status: 'skipped', + }, + ]; + render( + , + ); + + expect(positionedStyleFor('run-flow-node-request').width).toBe('188px'); + expect(positionedStyleFor('run-flow-node-news').width).toBe('244px'); + expect(nodeStyleFor('run-flow-node-news').width).toBe('244px'); + expect(positionedStyleFor('run-flow-node-save').width).toBe('188px'); + expect(nodeStyleFor('run-flow-node-save').width).toBe('188px'); + expect(positionedStyleFor('run-flow-node-notification').width).toBe('188px'); + expect(nodeStyleFor('run-flow-node-notification').width).toBe('188px'); + }); + + it('shows an inline expand control for expandable topology groups without selecting the node', () => { + const onSelectNode = vi.fn(); + const onToggleExpanded = vi.fn(); + render( + , + ); + + fireEvent.click(screen.getByRole('button', { name: '展开 新闻舆情 运行尝试' })); + + expect(screen.getByTestId('run-flow-node-topology_data_news_search')).toHaveClass('pb-8'); + expect(onToggleExpanded).toHaveBeenCalledWith('topology_data_news_search'); + expect(onSelectNode).not.toHaveBeenCalled(); + }); + + it('uses clearer default and selected card states without changing border width', () => { + render( + , + ); + + expect(screen.getByTestId('run-flow-node-request')).toHaveClass( + 'border-2', + 'border-subtle/80', + 'ring-1', + 'ring-white/5', + ); + expect(screen.getByTestId('run-flow-node-news')).toHaveClass( + 'border-2', + 'border-primary/85', + 'bg-primary/8', + 'ring-2', + 'ring-primary/25', + ); + }); + + it('emphasizes selected provider group paths including internal fallback attempts', () => { + render( + , + ); + + expect(screen.getByTestId('run-flow-edge-task-realtime')).toHaveAttribute('stroke-width', '2.4'); + expect(screen.getByTestId('run-flow-edge-task-realtime')).toHaveAttribute('opacity', '0.82'); + expect(screen.getByTestId('run-flow-edge-realtime-first')).toHaveAttribute('stroke-width', '3'); + expect(screen.getByTestId('run-flow-edge-realtime-first')).toHaveAttribute('opacity', '0.95'); + expect(screen.getByTestId('run-flow-edge-realtime-fallback')).toHaveAttribute('stroke-width', '3.5'); + expect(screen.getByTestId('run-flow-edge-realtime-fallback')).toHaveAttribute('opacity', '0.95'); + expect(screen.getByTestId('run-flow-edge-daily-llm')).toHaveAttribute('stroke-width', '1.75'); + expect(screen.getByTestId('run-flow-edge-daily-llm')).toHaveAttribute('opacity', '0.18'); + }); + + it('keeps expanded provider attempts grouped under their parent in compact cards', () => { + const onToggleExpanded = vi.fn(); + render( + , + ); + + expect(layoutRowFor('run-flow-node-provider_realtime_tushare_1')).toBe( + layoutRowFor('run-flow-node-topology_data_realtime_quote') + 1, + ); + expect(layoutRowFor('run-flow-node-provider_realtime_akshare_2')).toBe( + layoutRowFor('run-flow-node-provider_realtime_tushare_1') + 1, + ); + expect(layoutRowFor('run-flow-node-daily')).toBeGreaterThan( + layoutRowFor('run-flow-node-provider_realtime_akshare_2'), + ); + expect(topFor('run-flow-node-topology_data_realtime_quote')).toBeLessThan( + topFor('run-flow-node-provider_realtime_tushare_1'), + ); + expect(heightFor('run-flow-node-topology_data_realtime_quote')).toBe(112); + expect(heightFor('run-flow-node-provider_realtime_tushare_1')).toBe(96); + expect(topFor('run-flow-node-provider_realtime_tushare_1')).toBe( + topFor('run-flow-node-topology_data_realtime_quote') + 112 + 42, + ); + expect(topFor('run-flow-node-provider_realtime_akshare_2')).toBe( + topFor('run-flow-node-provider_realtime_tushare_1') + 96 + 42, + ); + expect(topFor('run-flow-node-daily')).toBe( + topFor('run-flow-node-provider_realtime_akshare_2') + 96 + 40, + ); + const groupBackground = screen.getByTestId('run-flow-expanded-group-topology_data_realtime_quote'); + const groupBackgroundTop = parseFloat(groupBackground.style.top); + const groupBackgroundBottom = groupBackgroundTop + parseFloat(groupBackground.style.height); + expect(groupBackgroundTop).toBe(topFor('run-flow-node-topology_data_realtime_quote') - 18); + expect(groupBackgroundBottom).toBe( + topFor('run-flow-node-provider_realtime_akshare_2') + heightFor('run-flow-node-provider_realtime_akshare_2') + 18, + ); + const canvasMinHeight = parseFloat((groupBackground.parentElement as HTMLElement).style.minHeight); + expect(canvasMinHeight).toBeGreaterThan(groupBackgroundBottom); + expect(screen.getByTestId('run-flow-node-provider_realtime_tushare_1')).toHaveClass( + 'bg-base/70', + 'shadow-none', + ); + expect(nodeStyleFor('run-flow-node-provider_realtime_tushare_1').width).toBe('244px'); + expect(nodeStyleFor('run-flow-node-provider_realtime_tushare_1').height).toBe('96px'); + const toggle = screen.getByTestId('run-flow-node-topology_data_realtime_quote-toggle'); + expect(toggle).toHaveClass('h-[18px]', 'gap-0.5', 'px-1', 'text-[9px]', 'leading-none'); + expect(toggle.querySelector('svg')).toHaveClass('h-2', 'w-2'); + }); + + it('keeps multiple expanded provider groups as non-interleaving layout blocks', () => { + render( + , + ); + + expect(layoutRowFor('run-flow-node-provider_realtime_tushare_1')).toBe( + layoutRowFor('run-flow-node-topology_data_realtime_quote') + 1, + ); + expect(layoutRowFor('run-flow-node-provider_realtime_akshare_2')).toBe( + layoutRowFor('run-flow-node-provider_realtime_tushare_1') + 1, + ); + expect(layoutRowFor('run-flow-node-topology_data_news_search')).toBeGreaterThan( + layoutRowFor('run-flow-node-provider_realtime_akshare_2'), + ); + expect(layoutRowFor('run-flow-node-provider_news_tavily_1')).toBe( + layoutRowFor('run-flow-node-topology_data_news_search') + 1, + ); + expect(layoutRowFor('run-flow-node-provider_news_searxng_2')).toBe( + layoutRowFor('run-flow-node-provider_news_tavily_1') + 1, + ); + expect(layoutRowFor('run-flow-node-daily')).toBeGreaterThan( + layoutRowFor('run-flow-node-provider_news_searxng_2'), + ); + expect(screen.getByTestId('run-flow-expanded-group-topology_data_realtime_quote')).toBeInTheDocument(); + expect(screen.getByTestId('run-flow-expanded-group-topology_data_news_search')).toBeInTheDocument(); + }); }); diff --git a/apps/dsa-web/src/components/run-flow/__tests__/RunFlowNodeDetails.test.tsx b/apps/dsa-web/src/components/run-flow/__tests__/RunFlowNodeDetails.test.tsx new file mode 100644 index 000000000..16a349287 --- /dev/null +++ b/apps/dsa-web/src/components/run-flow/__tests__/RunFlowNodeDetails.test.tsx @@ -0,0 +1,108 @@ +import { render, screen } from '@testing-library/react'; +import { describe, expect, it } from 'vitest'; +import type { RunFlowNode } from '../../../types/runFlow'; +import { RunFlowNodeDetails } from '../RunFlowNodeDetails'; + +describe('RunFlowNodeDetails', () => { + it('hides provider metrics that do not apply to queue nodes', () => { + const node: RunFlowNode = { + id: 'task_queue', + lane: 'entry', + kind: 'queue', + label: '任务队列', + status: 'success', + startedAt: '2026-06-08T22:14:25', + message: '任务进入运行队列', + }; + + render(); + + expect(screen.getByText('任务队列')).toBeInTheDocument(); + expect(screen.getByText('类型')).toBeInTheDocument(); + expect(screen.getByText('队列')).toBeInTheDocument(); + expect(screen.getByText('开始时间')).toBeInTheDocument(); + expect(screen.queryByText('提供方')).not.toBeInTheDocument(); + expect(screen.queryByText('耗时')).not.toBeInTheDocument(); + expect(screen.queryByText('尝试次数')).not.toBeInTheDocument(); + expect(screen.queryByText('记录数')).not.toBeInTheDocument(); + }); + + it('renders ContextPack quality metadata as structured details instead of raw JSON', () => { + const node: RunFlowNode = { + id: 'context_pack', + lane: 'analysis', + kind: 'analysis', + label: 'ContextPack', + status: 'degraded', + metadata: { + topologyGroup: 'context_pack', + packVersion: '1.0', + counts: { + available: 4, + missing: 1, + partial: 1, + fallback: 0, + }, + dataQuality: { + overallScore: 91, + level: 'good', + blockScores: { + quote: 100, + dailyBars: 100, + technical: 100, + news: 35, + }, + }, + context_status_counts: { + success: 4, + degraded: 1, + skipped: 1, + }, + }, + }; + + render(); + + expect(screen.getByText('上下文质量')).toBeInTheDocument(); + expect(screen.getByText('综合评分')).toBeInTheDocument(); + expect(screen.getByText('91')).toBeInTheDocument(); + expect(screen.getByText('数据块评分')).toBeInTheDocument(); + expect(screen.getByText('news')).toBeInTheDocument(); + expect(screen.getByText('35')).toBeInTheDocument(); + expect(screen.getByText('版本')).toBeInTheDocument(); + expect(screen.getByText('1.0')).toBeInTheDocument(); + expect(screen.queryByText('提供方')).not.toBeInTheDocument(); + expect(screen.queryByText('耗时')).not.toBeInTheDocument(); + expect(screen.queryByText('尝试次数')).not.toBeInTheDocument(); + expect(screen.queryByText('记录数')).not.toBeInTheDocument(); + expect(screen.queryByText('counts')).not.toBeInTheDocument(); + expect(screen.queryByText('dataQuality')).not.toBeInTheDocument(); + expect(screen.queryByText('context_status_counts')).not.toBeInTheDocument(); + expect(screen.queryByText(/overallScore/)).not.toBeInTheDocument(); + }); + + it('keeps provider metrics visible for data source nodes', () => { + const node: RunFlowNode = { + id: 'topology_data_realtime_quote', + lane: 'data_source', + kind: 'data_source', + label: '实时行情', + provider: 'TushareFetcher -> AkshareFetcher', + status: 'fallback', + durationMs: 750, + attempts: 2, + recordCount: 39, + }; + + render(); + + expect(screen.getByText('提供方')).toBeInTheDocument(); + expect(screen.getByText('TushareFetcher -> AkshareFetcher')).toBeInTheDocument(); + expect(screen.getByText('耗时')).toBeInTheDocument(); + expect(screen.getByText('750 ms')).toBeInTheDocument(); + expect(screen.getByText('尝试次数')).toBeInTheDocument(); + expect(screen.getByText('2')).toBeInTheDocument(); + expect(screen.getByText('记录数')).toBeInTheDocument(); + expect(screen.getByText('39')).toBeInTheDocument(); + }); +}); diff --git a/apps/dsa-web/src/components/run-flow/__tests__/RunFlowPanel.test.tsx b/apps/dsa-web/src/components/run-flow/__tests__/RunFlowPanel.test.tsx index a9613631c..9732f9b52 100644 --- a/apps/dsa-web/src/components/run-flow/__tests__/RunFlowPanel.test.tsx +++ b/apps/dsa-web/src/components/run-flow/__tests__/RunFlowPanel.test.tsx @@ -285,6 +285,8 @@ describe('RunFlowPanel', () => { expect(await screen.findByTestId('run-flow-panel')).toBeInTheDocument(); expect(screen.getByText('贵州茅台运行流')).toBeInTheDocument(); + expect(screen.getByTestId('run-flow-layout')).toHaveClass('xl:grid-cols-[minmax(0,1fr)_19.25rem]'); + expect(screen.getByTestId('run-flow-events-column')).toHaveClass('xl:max-h-[calc(100vh-18rem)]'); expect(screen.getByTestId('run-flow-graph')).toBeInTheDocument(); expect(screen.getByTestId('run-flow-events')).toBeInTheDocument(); expect(await screen.findByTestId('run-flow-node-details')).toHaveTextContent('新闻舆情'); @@ -294,7 +296,7 @@ describe('RunFlowPanel', () => { expect(screen.getByTestId('run-flow-node-details')).toHaveTextContent('LLM 生成'); expect(screen.getByTestId('run-flow-node-details')).toHaveTextContent('DeepSeek'); - fireEvent.click(screen.getByRole('button', { name: '新闻舆情 节点，状态 Fallback' })); + fireEvent.click(screen.getByRole('button', { name: '新闻舆情 节点，状态 降级回退' })); expect(screen.getByTestId('run-flow-node-details')).toHaveTextContent('fallbackFrom'); expect(screen.getByTestId('run-flow-node-details')).toHaveTextContent('Tushare'); @@ -332,7 +334,7 @@ describe('RunFlowPanel', () => { expect(await screen.findByTestId('run-flow-node-details')).toHaveTextContent('新闻舆情'); expect(screen.getByText('保存')).toBeInTheDocument(); - expect(screen.getByRole('button', { name: '新闻舆情 节点，状态 Fallback' })).toHaveAttribute('aria-pressed', 'false'); + expect(screen.getByRole('button', { name: '新闻舆情 节点，状态 降级回退' })).toHaveAttribute('aria-pressed', 'false'); }); it('expands provider attempt groups from node details', async () => { diff --git a/apps/dsa-web/src/components/run-flow/__tests__/topologyViewModel.test.ts b/apps/dsa-web/src/components/run-flow/__tests__/topologyViewModel.test.ts index 60c72470c..e75a44d00 100644 --- a/apps/dsa-web/src/components/run-flow/__tests__/topologyViewModel.test.ts +++ b/apps/dsa-web/src/components/run-flow/__tests__/topologyViewModel.test.ts @@ -144,6 +144,7 @@ describe('buildRunFlowTopologyModel', () => { }); expect(newsGroup?.metadata).toMatchObject({ topologyGroup: 'provider_attempts', + topologyRole: 'provider_group', data_type: 'news_search', success_count: 1, failed_count: 1, @@ -246,6 +247,61 @@ describe('buildRunFlowTopologyModel', () => { expect(model.events.find((event) => event.id === 'evt-normalized-block')?.nodeId).toBe('context_pack'); }); + it('keeps retry-only provider groups successful when every attempt succeeds', () => { + const retryOnlySnapshot: RunFlowSnapshot = { + ...baseSnapshot, + nodes: baseSnapshot.nodes.map((node) => ( + node.id === 'provider_news_search_tavily_1' + ? { ...node, status: 'success' as const } + : node + )), + edges: baseSnapshot.edges.map((edge) => ( + edge.id === 'news-1-news-2' + ? { ...edge, kind: 'retry' as const, status: 'success' as const } + : edge + )), + }; + + const model = buildRunFlowTopologyModel(retryOnlySnapshot); + const newsGroup = model.nodes.find((node) => node.id === 'topology_data_news_search'); + + expect(newsGroup).toMatchObject({ + status: 'success', + attempts: 2, + }); + expect(newsGroup?.metadata).toMatchObject({ + success_count: 2, + failed_count: 0, + fallback_count: 0, + retry_count: 1, + }); + }); + + it('marks mixed success and failure without recovery transitions as degraded', () => { + const degradedSnapshot: RunFlowSnapshot = { + ...baseSnapshot, + edges: baseSnapshot.edges.map((edge) => ( + edge.id === 'news-1-news-2' + ? { ...edge, kind: 'data' as const, status: 'success' as const } + : edge + )), + }; + + const model = buildRunFlowTopologyModel(degradedSnapshot); + const newsGroup = model.nodes.find((node) => node.id === 'topology_data_news_search'); + + expect(newsGroup).toMatchObject({ + status: 'degraded', + attempts: 2, + }); + expect(newsGroup?.metadata).toMatchObject({ + success_count: 1, + failed_count: 1, + fallback_count: 0, + retry_count: 0, + }); + }); + it('attaches context block states to ContextPack and remaps events', () => { const model = buildRunFlowTopologyModel(baseSnapshot); const contextPack = model.nodes.find((node) => node.id === 'context_pack'); @@ -285,4 +341,66 @@ describe('buildRunFlowTopologyModel', () => { ); expect(model.events.find((event) => event.id === 'evt-news-1')?.nodeId).toBe('provider_news_search_tavily_1'); }); + + it('adds stable topology metadata to expanded provider attempts even when data_type is missing', () => { + const snapshotWithoutDataType: RunFlowSnapshot = { + ...baseSnapshot, + nodes: baseSnapshot.nodes.map((node) => { + if (!node.id.startsWith('provider_')) { + return node; + } + return { + ...node, + metadata: {}, + }; + }), + }; + + const model = buildRunFlowTopologyModel(snapshotWithoutDataType, { + expandedGroupIds: new Set(['topology_data_news_search']), + }); + const tavily = model.nodes.find((node) => node.id === 'provider_news_search_tavily_1'); + const searxng = model.nodes.find((node) => node.id === 'provider_news_search_searxng_2'); + + expect(tavily?.metadata).toMatchObject({ + data_type: 'news_search', + topologyParentId: 'topology_data_news_search', + topologyRole: 'provider_attempt', + topologyOrder: 1, + }); + expect(searxng?.metadata).toMatchObject({ + data_type: 'news_search', + topologyParentId: 'topology_data_news_search', + topologyRole: 'provider_attempt', + topologyOrder: 2, + }); + }); + + it('keeps provider group running while any provider attempt is still running', () => { + const runningSnapshot: RunFlowSnapshot = { + ...baseSnapshot, + status: 'running', + nodes: baseSnapshot.nodes.map((node) => { + if (node.id === 'provider_news_search_tavily_1') { + return { + ...node, + status: 'success', + }; + } + if (node.id === 'provider_news_search_searxng_2') { + return { + ...node, + status: 'running', + endedAt: null, + }; + } + return node; + }), + }; + + const model = buildRunFlowTopologyModel(runningSnapshot); + const newsGroup = model.nodes.find((node) => node.id === 'topology_data_news_search'); + + expect(newsGroup?.status).toBe('running'); + }); }); diff --git a/apps/dsa-web/src/components/run-flow/topologyViewModel.ts b/apps/dsa-web/src/components/run-flow/topologyViewModel.ts index 1ed7beec9..1125a5227 100644 --- a/apps/dsa-web/src/components/run-flow/topologyViewModel.ts +++ b/apps/dsa-web/src/components/run-flow/topologyViewModel.ts @@ -106,11 +106,16 @@ const labelForDataType = (dataType: string, nodes: RunFlowNode[]): string => { const groupStatus = (nodes: RunFlowNode[], edges: RunFlowEdge[]): RunFlowStatus => { const statuses = nodes.map((node) => node.status); + if (statuses.length === 0) return 'unknown'; + if (statuses.includes('cancel_requested')) return 'cancel_requested'; + if (statuses.some((status) => status === 'running' || status === 'pending')) return 'running'; + if (statuses.every((status) => status === 'success')) return 'success'; const hasSuccess = statuses.includes('success'); - const hasFallbackSignal = statuses.includes('fallback') - || edges.some((edge) => edge.kind === 'fallback' || edge.kind === 'retry'); - if (hasSuccess && hasFallbackSignal) return 'fallback'; - if (hasSuccess && statuses.some((status) => ['failed', 'timeout', 'degraded'].includes(status))) return 'degraded'; + const hasFailedOrTimeout = statuses.some((status) => status === 'failed' || status === 'timeout'); + const hasFallbackAttempt = statuses.includes('fallback'); + const hasRecoveryTransition = edges.some((edge) => edge.kind === 'fallback' || edge.kind === 'retry'); + if (hasSuccess && (hasFailedOrTimeout || hasFallbackAttempt) && hasRecoveryTransition) return 'fallback'; + if (hasSuccess && statuses.some((status) => ['failed', 'timeout', 'degraded', 'fallback'].includes(status))) return 'degraded'; return statuses.reduce((winner, status) => ( statusRank(status) > statusRank(winner) ? status : winner ), 'unknown'); @@ -159,6 +164,7 @@ const buildProviderGroupNode = ( recordCount: firstDefinedRecordCount([...sortedAttempts].reverse()), metadata: { topologyGroup: 'provider_attempts', + topologyRole: 'provider_group', expanded, data_type: dataType, provider_chain: providerChain, @@ -236,6 +242,25 @@ export const buildRunFlowTopologyModel = ( nodeIdMap.set(node.id, options.expandedGroupIds?.has(groupId) ? node.id : groupId); }); + const attemptTopologyById = new Map(); + attemptsByDataType.forEach((attempts, dataType) => { + [...attempts] + .sort((left, right) => ( + (nodeTime(left) ?? Number.MAX_SAFE_INTEGER) - (nodeTime(right) ?? Number.MAX_SAFE_INTEGER) + )) + .forEach((node, index) => { + attemptTopologyById.set(node.id, { + dataType, + groupId: `${PROVIDER_GROUP_PREFIX}${dataType}`, + order: index + 1, + }); + }); + }); + const providerGroupNodes = Array.from(attemptsByDataType.entries()).map(([dataType, attempts]) => { const attemptIds = new Set(attempts.map((node) => node.id)); const attemptEdges = snapshot.edges.filter((edge) => attemptIds.has(edge.from) && attemptIds.has(edge.to)); @@ -256,6 +281,22 @@ export const buildRunFlowTopologyModel = ( const visibleNodes = snapshot.nodes .filter((node) => !collapsedNodeIds.has(node.id)) + .map((node) => { + const topology = attemptTopologyById.get(node.id); + if (!topology) { + return node; + } + return { + ...node, + metadata: { + ...(node.metadata || {}), + data_type: topology.dataType, + topologyParentId: topology.groupId, + topologyRole: 'provider_attempt', + topologyOrder: topology.order, + }, + }; + }) .map((node) => attachContextBlocksToPack(node, contextBlocks)); const nodes = [...visibleNodes, ...providerGroupNodes]; diff --git a/apps/dsa-web/src/components/tasks/TaskPanel.tsx b/apps/dsa-web/src/components/tasks/TaskPanel.tsx index 5d0d33a3f..fbdbbb26c 100644 --- a/apps/dsa-web/src/components/tasks/TaskPanel.tsx +++ b/apps/dsa-web/src/components/tasks/TaskPanel.tsx @@ -36,102 +36,110 @@ const TaskItem: React.FC = ({ task, onOpenRunFlow }) => { const requestedPhaseVariant = task.analysisPhase === 'auto' ? 'default' : 'info'; return ( -
- {/* 状态图标 */} -
- {isProcessing ? ( - - ) : isCancelRequested ? ( - - ) : isPending ? ( - - ) : null} -
+
+
+
+
+ {isProcessing ? ( + + ) : isCancelRequested ? ( + + ) : isPending ? ( + + ) : null} +
- {/* 任务信息 */} -
-
- - {task.stockName || task.stockCode} - - - {task.stockCode} - -
- {task.message && ( -

- {task.message} -

- )} - {requestedPhaseLabel ? ( -
- - {requestedPhaseLabel} - -
- ) : null} -
-
-
-
- - {progress}% - -
- {traceId ? ( -
- - {t('taskPanel.diagnostics')} - - {traceId.length > 18 ? `${traceId.slice(0, 10)}...` : traceId} +
+
+ + {task.stockName || task.stockCode} + + + {task.stockCode} -
-
- Trace: - - {traceId} -
-
- ) : null} +
+
+ +
+ {onOpenRunFlow ? ( + + + + + + ) : null} + + + {statusLabel} + +
- {/* 状态标签 */} -
- {onOpenRunFlow ? ( - - - - - - ) : null} - - - {statusLabel} - + {task.message ? ( +

+ {task.message} +

+ ) : null} + + {requestedPhaseLabel ? ( +
+ + {requestedPhaseLabel} + +
+ ) : null} + +
+
+
+
+ + {progress}% +
+ + {traceId ? ( +
+ + {t('taskPanel.diagnostics')} + + {traceId.length > 18 ? `${traceId.slice(0, 10)}...` : traceId} + + +
+ Trace: + + {traceId} + +
+
+ ) : null}
); }; diff --git a/apps/dsa-web/src/components/tasks/__tests__/TaskPanel.test.tsx b/apps/dsa-web/src/components/tasks/__tests__/TaskPanel.test.tsx index 31ed23bd8..a324bfe50 100644 --- a/apps/dsa-web/src/components/tasks/__tests__/TaskPanel.test.tsx +++ b/apps/dsa-web/src/components/tasks/__tests__/TaskPanel.test.tsx @@ -72,6 +72,38 @@ describe('TaskPanel', () => { expect(container.querySelector('.home-subpanel')).toBeTruthy(); }); + it('keeps narrow sidebar task metadata in rows instead of squeezing diagnostics vertically', () => { + render( + , + ); + + const item = screen.getByTestId('task-panel-item'); + expect(item).toHaveClass('grid'); + expect(item).not.toHaveClass('flex'); + expect(screen.getByText('长飞光纤')).toHaveClass('truncate'); + expect(screen.getByText('601869.SH')).toHaveClass('shrink-0'); + expect(screen.getByText('32%')).toBeInTheDocument(); + + const diagnosticsSummary = screen.getByTestId('task-panel-diagnostics-summary'); + expect(diagnosticsSummary).toHaveClass('grid-cols-[auto_minmax(0,1fr)_auto]'); + expect(screen.getByText('运行诊断')).toHaveClass('whitespace-nowrap'); + expect(screen.getByText('c5b9665a64...')).toHaveClass('truncate'); + expect(screen.getByRole('button', { name: '查看 长飞光纤 运行流' })).toBeInTheDocument(); + }); + it('opens the run-flow view from an active task icon button', () => { const onOpenRunFlow = vi.fn(); render( diff --git a/apps/dsa-web/src/hooks/__tests__/useRunFlowSnapshot.test.tsx b/apps/dsa-web/src/hooks/__tests__/useRunFlowSnapshot.test.tsx index a1fa0719f..0b5856bb6 100644 --- a/apps/dsa-web/src/hooks/__tests__/useRunFlowSnapshot.test.tsx +++ b/apps/dsa-web/src/hooks/__tests__/useRunFlowSnapshot.test.tsx @@ -140,6 +140,92 @@ describe('useRunFlowSnapshot', () => { await waitFor(() => expect(analysisApi.getTaskFlow).toHaveBeenCalledTimes(2)); }); + it('updates started live flow nodes in place when finish events arrive', async () => { + vi.mocked(analysisApi.getTaskFlow).mockResolvedValue(snapshot); + + const { result } = renderHook(() => useRunFlowSnapshot({ + source: { type: 'task', taskId: 'task-1' }, + enabled: true, + })); + + await waitFor(() => expect(result.current.snapshot).not.toBeNull()); + + act(() => { + taskStreamCalls.at(-1)?.onTaskFlowEvent?.( + { + taskId: 'task-1', + stockCode: '600519', + status: 'processing', + progress: 30, + reportType: 'detailed', + createdAt: '2026-06-08T08:00:00Z', + }, + { + id: 'flow-provider-started', + timestamp: '2026-06-08T08:00:01Z', + severity: 'info', + type: 'provider_run_started', + nodeId: 'provider_daily_data_dailyfetcher_1', + title: '日线K线开始', + metadata: { + provider: 'DailyFetcher', + dataType: 'daily_data', + node: { + id: 'provider_daily_data_dailyfetcher_1', + lane: 'data_source', + kind: 'data_source', + label: '日线K线 · DailyFetcher', + status: 'running', + provider: 'DailyFetcher', + metadata: { dataType: 'daily_data' }, + }, + }, + }, + ); + taskStreamCalls.at(-1)?.onTaskFlowEvent?.( + { + taskId: 'task-1', + stockCode: '600519', + status: 'processing', + progress: 35, + reportType: 'detailed', + createdAt: '2026-06-08T08:00:00Z', + }, + { + id: 'flow-provider-finished', + timestamp: '2026-06-08T08:00:02Z', + severity: 'success', + type: 'provider_run', + nodeId: 'provider_daily_data_dailyfetcher_1', + title: '日线K线成功', + metadata: { + provider: 'DailyFetcher', + dataType: 'daily_data', + node: { + id: 'provider_daily_data_dailyfetcher_1', + lane: 'data_source', + kind: 'data_source', + label: '日线K线 · DailyFetcher', + status: 'success', + provider: 'DailyFetcher', + recordCount: 30, + metadata: { dataType: 'daily_data' }, + }, + }, + }, + ); + }); + + const providerNodes = result.current.snapshot?.nodes.filter((node) => ( + node.id === 'provider_daily_data_dailyfetcher_1' + )); + expect(providerNodes).toHaveLength(1); + expect(providerNodes?.[0]).toEqual(expect.objectContaining({ + status: 'success', + recordCount: 30, + })); + }); + it('does not enable task stream for history snapshots', async () => { vi.mocked(historyApi.getRecordFlow).mockResolvedValue({ ...snapshot, status: 'success' }); @@ -565,4 +651,100 @@ describe('useRunFlowSnapshot', () => { expect(lateEvent?.metadata).not.toHaveProperty('node'); expect(result.current.snapshot?.nodes.some((node) => node.id === 'provider_news_1')).toBe(true); }); + + it('skips replaying buffered events when refreshed snapshot already has a completed node', async () => { + const initialRequest = createDeferred(); + const refreshedRequest = createDeferred(); + vi.mocked(analysisApi.getTaskFlow) + .mockReturnValueOnce(initialRequest.promise) + .mockReturnValueOnce(refreshedRequest.promise); + + const { result } = renderHook(() => useRunFlowSnapshot({ + source: { type: 'task', taskId: 'task-1' }, + enabled: true, + })); + + act(() => { + initialRequest.resolve(snapshot); + }); + + await waitFor(() => expect(result.current.snapshot?.events).toHaveLength(1)); + + const liveNotificationEvent = { + id: 'flow-live-notification', + timestamp: '2026-06-08T08:00:02Z', + severity: 'warning' as const, + type: 'notification_run' as const, + nodeId: 'notification_report_1', + title: '通知跳过', + metadata: { + channel: 'report', + node: { + id: 'notification_report_1', + lane: 'artifact', + kind: 'notification', + label: '推送通知 · report', + status: 'skipped', + }, + }, + }; + + act(() => { + taskStreamCalls.at(-1)?.onTaskFlowEvent?.( + { + taskId: 'task-1', + stockCode: '600519', + status: 'processing', + progress: 99, + reportType: 'detailed', + createdAt: '2026-06-08T08:00:00Z', + }, + liveNotificationEvent, + ); + taskStreamCalls.at(-1)?.onTaskCompleted?.({ + taskId: 'task-1', + stockCode: '600519', + status: 'completed', + progress: 100, + reportType: 'detailed', + createdAt: '2026-06-08T08:00:00Z', + }); + }); + + await waitFor(() => expect(analysisApi.getTaskFlow).toHaveBeenCalledTimes(2)); + + act(() => { + refreshedRequest.resolve({ + ...snapshot, + status: 'success', + nodes: [ + ...snapshot.nodes, + { + id: 'notification_report_1', + lane: 'artifact', + kind: 'notification', + label: '推送通知 · report', + status: 'skipped', + }, + ], + events: [ + ...snapshot.events, + { + id: 'history-notification', + timestamp: '2026-06-08T08:00:02Z', + severity: 'warning', + type: 'notification_run', + nodeId: 'notification_report_1', + title: '通知跳过', + }, + ], + }); + }); + + await waitFor(() => expect(result.current.snapshot?.status).toBe('success')); + + expect(result.current.snapshot?.events.some((event) => event.id === 'history-notification')).toBe(true); + expect(result.current.snapshot?.events.some((event) => event.id === 'flow-live-notification')).toBe(false); + expect(result.current.snapshot?.nodes.filter((node) => node.id === 'notification_report_1')).toHaveLength(1); + }); }); diff --git a/apps/dsa-web/src/hooks/useRunFlowSnapshot.ts b/apps/dsa-web/src/hooks/useRunFlowSnapshot.ts index 4220a4303..0aa7b0ed8 100644 --- a/apps/dsa-web/src/hooks/useRunFlowSnapshot.ts +++ b/apps/dsa-web/src/hooks/useRunFlowSnapshot.ts @@ -149,6 +149,28 @@ const appendEdge = ( ]; }; +const refreshIncomingEdgeStatus = ( + edges: RunFlowEdge[], + nodeId: string | null, + status?: RunFlowEdge['status'], +): RunFlowEdge[] => { + if (!nodeId || !status) { + return edges; + } + let changed = false; + const refreshed = edges.map((edge) => { + if (edge.to !== nodeId || edge.status === status) { + return edge; + } + changed = true; + return { + ...edge, + status, + }; + }); + return changed ? refreshed : edges; +}; + const providerTransitionKind = ( previous: { provider: string | null; success: boolean; fallbackTo: string | null }, current: { provider: string | null; success: boolean; fallbackFrom: string | null }, @@ -191,12 +213,16 @@ const appendDerivedEdge = ( return edges; } - if (displayEvent.type === 'provider_run') { + if (displayEvent.type === 'provider_run' || displayEvent.type === 'provider_run_started') { const dataType = dataTypeFromEvent(displayEvent, node); const currentTime = eventTime(displayEvent); const previousEvent = events .filter((event) => { - if (event.id === displayEvent.id || event.type !== 'provider_run' || !event.nodeId) { + if ( + event.id === displayEvent.id + || (event.type !== 'provider_run' && event.type !== 'provider_run_started') + || !event.nodeId + ) { return false; } if (eventTime(event) >= currentTime) { @@ -231,7 +257,7 @@ const appendDerivedEdge = ( return appendEdge(edges, previousNode.id, nodeId, transitionKind, node.status, label, message); } - if (displayEvent.type === 'llm_run') { + if (displayEvent.type === 'llm_run' || displayEvent.type === 'llm_run_started') { const anchor = nodeById.has('analysis_pipeline') ? 'analysis_pipeline' : 'task_queue'; return nodeById.has(anchor) ? appendEdge(edges, anchor, nodeId, 'data', node.status, '生成') @@ -239,7 +265,7 @@ const appendDerivedEdge = ( } if (displayEvent.type === 'history_run') { - const anchor = latestEventNodeId(events, nodeById, ['llm_run'], displayEvent) + const anchor = latestEventNodeId(events, nodeById, ['llm_run', 'llm_run_started'], displayEvent) || (nodeById.has('analysis_pipeline') ? 'analysis_pipeline' : 'task_queue'); return nodeById.has(anchor) ? appendEdge(edges, anchor, nodeId, 'data', node.status, '保存') @@ -248,7 +274,7 @@ const appendDerivedEdge = ( if (displayEvent.type === 'notification_run') { const anchor = latestEventNodeId(events, nodeById, ['history_run'], displayEvent) - || latestEventNodeId(events, nodeById, ['llm_run'], displayEvent) + || latestEventNodeId(events, nodeById, ['llm_run', 'llm_run_started'], displayEvent) || (nodeById.has('analysis_pipeline') ? 'analysis_pipeline' : 'task_queue'); return nodeById.has(anchor) ? appendEdge(edges, anchor, nodeId, 'control', node.status, '通知') @@ -331,12 +357,16 @@ const mergeFlowEventIntoSnapshot = ( const nodes = upsertNode(snapshot.nodes, node); const edges = eventAlreadyPresent ? snapshot.edges - : appendDerivedEdge( - nodes, - snapshot.edges, - events, - displayEvent, + : refreshIncomingEdgeStatus( + appendDerivedEdge( + nodes, + snapshot.edges, + events, + displayEvent, + eventNodeId(displayEvent, node), + ), eventNodeId(displayEvent, node), + node?.status, ); return { @@ -359,11 +389,34 @@ const rememberFlowEvent = (events: RunFlowEvent[], flowEvent: RunFlowEvent): Run .slice(-MAX_BUFFERED_FLOW_EVENTS); }; +const ACTIVE_NODE_STATUSES = new Set(['pending', 'running', 'cancel_requested']); + +const replayEventNodeId = (flowEvent: RunFlowEvent): string | null => { + const nodeCandidate = flowEvent.metadata?.node; + if (isRunFlowNode(nodeCandidate)) { + return nodeCandidate.id; + } + return flowEvent.nodeId || null; +}; + +const shouldReplayFlowEvent = (snapshot: RunFlowSnapshot, flowEvent: RunFlowEvent): boolean => { + const nodeId = replayEventNodeId(flowEvent); + if (!nodeId) { + return true; + } + const existingNode = snapshot.nodes.find((node) => node.id === nodeId); + return !existingNode || ACTIVE_NODE_STATUSES.has(existingNode.status); +}; + const replayFlowEvents = ( snapshot: RunFlowSnapshot, flowEvents: RunFlowEvent[], ): RunFlowSnapshot => flowEvents.reduce( - (currentSnapshot, flowEvent) => mergeFlowEventIntoSnapshot(currentSnapshot, flowEvent), + (currentSnapshot, flowEvent) => ( + shouldReplayFlowEvent(currentSnapshot, flowEvent) + ? mergeFlowEventIntoSnapshot(currentSnapshot, flowEvent) + : currentSnapshot + ), snapshot, ); diff --git a/apps/dsa-web/src/i18n/uiText.ts b/apps/dsa-web/src/i18n/uiText.ts index 786d2005d..7b78411ee 100644 --- a/apps/dsa-web/src/i18n/uiText.ts +++ b/apps/dsa-web/src/i18n/uiText.ts @@ -244,8 +244,8 @@ const zh = { 'runFlow.status.running': '运行中', 'runFlow.status.success': '成功', 'runFlow.status.failed': '失败', - 'runFlow.status.degraded': '降级', - 'runFlow.status.fallback': 'Fallback', + 'runFlow.status.degraded': '部分降级', + 'runFlow.status.fallback': '降级回退', 'runFlow.status.timeout': '超时', 'runFlow.status.cancelRequested': '请求取消', 'runFlow.status.cancelled': '已取消', @@ -257,8 +257,10 @@ const zh = { 'runFlow.severity.danger': '危险', 'runFlow.edge.data': '数据', 'runFlow.edge.control': '控制', - 'runFlow.edge.fallback': '降级', + 'runFlow.edge.fallback': '降级回退', 'runFlow.edge.retry': '重试', + 'runFlow.edgeLabel.invoke': '调用', + 'runFlow.edgeLabel.details': '详情', 'runFlow.nodeKind.entry': '入口', 'runFlow.nodeKind.queue': '队列', 'runFlow.nodeKind.dataSource': '数据源', @@ -267,7 +269,7 @@ const zh = { 'runFlow.nodeKind.artifact': '产物', 'runFlow.nodeKind.notification': '通知', 'runFlow.summary.elapsed': '总耗时', - 'runFlow.summary.fallbackCount': '降级/重试', + 'runFlow.summary.fallbackCount': '降级回退/重试', 'runFlow.summary.failedAttempts': '失败尝试', 'runFlow.summary.dataSources': '数据源', 'runFlow.summary.task': 'Task', @@ -278,13 +280,17 @@ const zh = { 'runFlow.graph.description': '自动分层展示入口、数据来源、分析引擎和产物链路。', 'runFlow.graph.nodeAria': '{label} 节点，状态 {status}', 'runFlow.graph.startedAt': '开始', + 'runFlow.graph.expand': '展开', + 'runFlow.graph.collapse': '收起', + 'runFlow.graph.expandNode': '展开 {label} 运行尝试', + 'runFlow.graph.collapseNode': '收起 {label} 运行尝试', 'runFlow.events.title': '事件流', 'runFlow.events.count': '{count} 条事件', 'runFlow.events.filters': '事件筛选', 'runFlow.events.filter.all': '全部', 'runFlow.events.filter.important': '关键', 'runFlow.events.filter.problems': '失败/告警', - 'runFlow.events.filter.fallback': '降级/重试', + 'runFlow.events.filter.fallback': '降级回退/重试', 'runFlow.events.filter.cancelled': '取消', 'runFlow.events.openNode': '查看事件 {title} 关联节点', 'runFlow.events.empty': '当前筛选下暂无事件。', @@ -292,7 +298,8 @@ const zh = { 'runFlow.nodeDetails.title': '节点详情', 'runFlow.nodeDetails.close': '关闭节点详情', 'runFlow.nodeDetails.kind': '类型', - 'runFlow.nodeDetails.provider': 'Provider', + 'runFlow.nodeDetails.version': '版本', + 'runFlow.nodeDetails.provider': '提供方', 'runFlow.nodeDetails.duration': '耗时', 'runFlow.nodeDetails.attempts': '尝试次数', 'runFlow.nodeDetails.recordCount': '记录数', @@ -308,6 +315,16 @@ const zh = { 'runFlow.nodeDetails.column.duration': '耗时', 'runFlow.nodeDetails.column.records': '记录', 'runFlow.nodeDetails.column.time': '时间', + 'runFlow.nodeDetails.contextQuality': '上下文质量', + 'runFlow.nodeDetails.overallScore': '综合评分', + 'runFlow.nodeDetails.qualityLevel': '等级', + 'runFlow.nodeDetails.blockScores': '数据块评分', + 'runFlow.nodeDetails.count.available': '可用', + 'runFlow.nodeDetails.count.missing': '缺失', + 'runFlow.nodeDetails.count.partial': '部分', + 'runFlow.nodeDetails.count.degraded': '部分降级', + 'runFlow.nodeDetails.count.fallback': '降级回退', + 'runFlow.nodeDetails.count.skipped': '跳过', 'report.addToWatchlist': '加入自选', 'report.removeFromWatchlist': '从自选删除', @@ -750,6 +767,8 @@ const en: Record = { 'runFlow.edge.control': 'Control', 'runFlow.edge.fallback': 'Fallback', 'runFlow.edge.retry': 'Retry', + 'runFlow.edgeLabel.invoke': 'Invoke', + 'runFlow.edgeLabel.details': 'Details', 'runFlow.nodeKind.entry': 'Entry', 'runFlow.nodeKind.queue': 'Queue', 'runFlow.nodeKind.dataSource': 'Data source', @@ -769,6 +788,10 @@ const en: Record = { 'runFlow.graph.description': 'Auto-layered lanes show entry, data sources, analysis engines, and artifact paths.', 'runFlow.graph.nodeAria': '{label} node, status {status}', 'runFlow.graph.startedAt': 'Start', + 'runFlow.graph.expand': 'Expand', + 'runFlow.graph.collapse': 'Collapse', + 'runFlow.graph.expandNode': 'Expand {label} attempts', + 'runFlow.graph.collapseNode': 'Collapse {label} attempts', 'runFlow.events.title': 'Event stream', 'runFlow.events.count': '{count} events', 'runFlow.events.filters': 'Event filters', @@ -783,6 +806,7 @@ const en: Record = { 'runFlow.nodeDetails.title': 'Node details', 'runFlow.nodeDetails.close': 'Close node details', 'runFlow.nodeDetails.kind': 'Kind', + 'runFlow.nodeDetails.version': 'Version', 'runFlow.nodeDetails.provider': 'Provider', 'runFlow.nodeDetails.duration': 'Duration', 'runFlow.nodeDetails.attempts': 'Attempts', @@ -799,6 +823,16 @@ const en: Record = { 'runFlow.nodeDetails.column.duration': 'Duration', 'runFlow.nodeDetails.column.records': 'Records', 'runFlow.nodeDetails.column.time': 'Time', + 'runFlow.nodeDetails.contextQuality': 'Context quality', + 'runFlow.nodeDetails.overallScore': 'Overall score', + 'runFlow.nodeDetails.qualityLevel': 'Level', + 'runFlow.nodeDetails.blockScores': 'Block scores', + 'runFlow.nodeDetails.count.available': 'Available', + 'runFlow.nodeDetails.count.missing': 'Missing', + 'runFlow.nodeDetails.count.partial': 'Partial', + 'runFlow.nodeDetails.count.degraded': 'Degraded', + 'runFlow.nodeDetails.count.fallback': 'Fallback', + 'runFlow.nodeDetails.count.skipped': 'Skipped', 'report.addToWatchlist': 'Add to watchlist', 'report.removeFromWatchlist': 'Remove from watchlist', diff --git a/data_provider/base.py b/data_provider/base.py index 780d2e90e..9813eacd1 100644 --- a/data_provider/base.py +++ b/data_provider/base.py @@ -26,7 +26,7 @@ import pandas as pd import numpy as np from src.data.stock_index_loader import get_index_stock_name from src.data.stock_mapping import STOCK_NAME_MAP, is_meaningful_stock_name -from src.services.run_diagnostics import record_provider_run +from src.services.run_diagnostics import record_provider_run, record_provider_run_started from .fundamental_adapter import AkshareFundamentalAdapter from .yfinance_fundamental_adapter import YfinanceFundamentalAdapter @@ -1206,6 +1206,11 @@ class DataFetcherManager: f"[数据源尝试 {attempt}/{total_fetchers}] [{fetcher.name}] " f"{market_label} {stock_code} {role}路由..." ) + record_provider_run_started( + data_type="daily_data", + provider=fetcher.name, + operation="get_daily_data", + ) df = self._call_fetcher_method( fetcher, "get_daily_data", @@ -1273,6 +1278,11 @@ class DataFetcherManager: fallback_to = fetchers[attempt].name if attempt < total_fetchers else None try: logger.info(f"[数据源尝试 {attempt}/{total_fetchers}] [{fetcher.name}] 获取 {stock_code}...") + record_provider_run_started( + data_type="daily_data", + provider=fetcher.name, + operation="get_daily_data", + ) df = self._call_fetcher_method( fetcher, "get_daily_data", @@ -1638,26 +1648,51 @@ class DataFetcherManager: if source == "efinance": fetcher = self._get_fetcher_by_name("EfinanceFetcher", capability="realtime_quote") if fetcher is not None and hasattr(fetcher, 'get_realtime_quote'): + record_provider_run_started( + data_type="realtime_quote", + provider=fetcher.name, + operation="get_realtime_quote", + ) quote = self._call_fetcher_method(fetcher, 'get_realtime_quote', stock_code) elif source == "akshare_em": fetcher = self._get_fetcher_by_name("AkshareFetcher", capability="realtime_quote") if fetcher is not None and hasattr(fetcher, 'get_realtime_quote'): + record_provider_run_started( + data_type="realtime_quote", + provider=fetcher.name, + operation="get_realtime_quote", + ) quote = self._call_fetcher_method(fetcher, 'get_realtime_quote', stock_code, source="em") elif source == "akshare_sina": fetcher = self._get_fetcher_by_name("AkshareFetcher", capability="realtime_quote") if fetcher is not None and hasattr(fetcher, 'get_realtime_quote'): + record_provider_run_started( + data_type="realtime_quote", + provider=fetcher.name, + operation="get_realtime_quote", + ) quote = self._call_fetcher_method(fetcher, 'get_realtime_quote', stock_code, source="sina") elif source in ("tencent", "akshare_qq"): fetcher = self._get_fetcher_by_name("AkshareFetcher", capability="realtime_quote") if fetcher is not None and hasattr(fetcher, 'get_realtime_quote'): + record_provider_run_started( + data_type="realtime_quote", + provider=fetcher.name, + operation="get_realtime_quote", + ) quote = self._call_fetcher_method(fetcher, 'get_realtime_quote', stock_code, source="tencent") elif source == "tushare": fetcher = self._get_fetcher_by_name("TushareFetcher", capability="realtime_quote") if fetcher is not None and hasattr(fetcher, 'get_realtime_quote'): + record_provider_run_started( + data_type="realtime_quote", + provider=fetcher.name, + operation="get_realtime_quote", + ) quote = self._call_fetcher_method(fetcher, 'get_realtime_quote', raw_stock_code or stock_code) provider_name = fetcher.name if fetcher is not None else source @@ -1807,6 +1842,11 @@ class DataFetcherManager: return None attempt_start = time.time() try: + record_provider_run_started( + data_type="realtime_quote", + provider=fetcher.name, + operation="get_realtime_quote", + ) q = self._call_fetcher_method(fetcher, 'get_realtime_quote', stock_code, **kw) if q is not None and q.has_basic_data(): record_provider_run( @@ -1901,12 +1941,13 @@ class DataFetcherManager: circuit_breaker = get_chip_circuit_breaker() + candidate_fetchers = [] # 直接遍历管理器已经按 priority 排好序的数据源列表 for fetcher in self._get_fetchers_snapshot(): # 只处理实现了筹码分布逻辑的数据源 if not hasattr(fetcher, 'get_chip_distribution'): continue - + fetcher_name = fetcher.name # 动态生成熔断器的 key，例如 "TushareFetcher" -> "tushare_chip" source_key = f"{fetcher_name.replace('Fetcher', '').lower()}_chip" @@ -1916,13 +1957,47 @@ class DataFetcherManager: logger.debug(f"[熔断] {fetcher_name} 筹码接口处于熔断状态，尝试下一个") continue + candidate_fetchers.append((fetcher, fetcher_name, source_key)) + + for index, (fetcher, fetcher_name, source_key) in enumerate(candidate_fetchers): + fallback_to = ( + candidate_fetchers[index + 1][1] + if index + 1 < len(candidate_fetchers) + else None + ) + attempt_start = time.time() try: + record_provider_run_started( + data_type="chip", + provider=fetcher_name, + operation="get_chip_distribution", + ) chip = self._call_fetcher_method(fetcher, 'get_chip_distribution', stock_code) + latency_ms = int((time.time() - attempt_start) * 1000) if _is_meaningful_chip_distribution(chip): + record_provider_run( + data_type="chip", + provider=fetcher_name, + operation="get_chip_distribution", + success=True, + latency_ms=latency_ms, + record_count=1, + ) circuit_breaker.record_success(source_key) logger.info(f"[筹码分布] {stock_code} 成功获取 (来源: {fetcher_name})") return chip else: + record_provider_run( + data_type="chip", + provider=fetcher_name, + operation="get_chip_distribution", + success=False, + latency_ms=latency_ms, + error_type="empty", + error_message="empty or incomplete chip distribution", + fallback_to=fallback_to, + record_count=0, + ) if chip is not None: logger.warning( "[筹码分布] %s 返回字段不完整或占位值，继续尝试下一个数据源", @@ -1931,6 +2006,17 @@ class DataFetcherManager: # 空结果或占位结果：释放 HALF_OPEN 探测名额，避免卡死 circuit_breaker.record_inconclusive(source_key) except Exception as e: + error_type, error_reason = summarize_exception(e) + record_provider_run( + data_type="chip", + provider=fetcher_name, + operation="get_chip_distribution", + success=False, + latency_ms=int((time.time() - attempt_start) * 1000), + error_type=error_type, + error_message=error_reason, + fallback_to=fallback_to, + ) logger.warning(f"[筹码分布] {fetcher_name} 获取 {stock_code} 失败: {e}") circuit_breaker.record_failure(source_key, str(e)) continue @@ -2017,16 +2103,60 @@ class DataFetcherManager: stock_code = normalize_stock_code(stock_code) if _market_tag(stock_code) != "cn": return [] - for fetcher in self._fetchers: - if not hasattr(fetcher, "get_belong_board"): - continue + candidate_fetchers = [ + fetcher + for fetcher in self._fetchers + if hasattr(fetcher, "get_belong_board") + ] + for index, fetcher in enumerate(candidate_fetchers): + fallback_to = ( + candidate_fetchers[index + 1].name + if index + 1 < len(candidate_fetchers) + else None + ) + start = time.time() try: + record_provider_run_started( + data_type="belong_boards", + provider=fetcher.name, + operation="get_belong_board", + ) raw_data = fetcher.get_belong_board(stock_code) boards = self._normalize_belong_boards(raw_data) if boards: + record_provider_run( + data_type="belong_boards", + provider=fetcher.name, + operation="get_belong_board", + success=True, + latency_ms=int((time.time() - start) * 1000), + record_count=len(boards), + ) logger.info(f"[{fetcher.name}] 获取所属板块成功: {stock_code}, count={len(boards)}") return boards + record_provider_run( + data_type="belong_boards", + provider=fetcher.name, + operation="get_belong_board", + success=False, + latency_ms=int((time.time() - start) * 1000), + error_type="empty", + error_message="empty belong boards", + fallback_to=fallback_to, + record_count=0, + ) except Exception as e: + error_type, error_reason = summarize_exception(e) + record_provider_run( + data_type="belong_boards", + provider=fetcher.name, + operation="get_belong_board", + success=False, + latency_ms=int((time.time() - start) * 1000), + error_type=error_type, + error_message=error_reason, + fallback_to=fallback_to, + ) logger.debug(f"[{fetcher.name}] 获取所属板块失败: {e}") continue return [] diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 057bb60bf..be4e3f30b 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -9,6 +9,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +- [修复] 问股从历史报告进入后的追问会持续携带当前标的，切回或重载已有会话时可从历史消息恢复基础当前标的，并由后端阻断未明确切换时的错误股票工具调用、交易所片段和指标缩写误路由。 +- [修复] 自选股加入和删除按等价股票代码匹配港股及大小写美股变体，避免 `00700`、`HK00700`、`00700.HK` 或 `aapl`、`AAPL` 被误判为不同标的。 +- [改进] #1390 P0 为个股分析与历史/回测展示新增可选八态 `action` / `action_label` 建议动作字段，保留 `operation_advice` 自由文本和 `decision_type=buy|hold|sell` 统计口径，不新增迁移或配置项。 +- [新功能] #1390 P1 新增独立 `DecisionSignal` 存储、Repository、Service 与 `/api/v1/decision-signals` API，支持按来源类型/市场/股票/动作/期限/阶段去重、按 `source_report_id` / `trace_id` 查询、同源过期信号续期且保留来源身份字段、禁止 expired 直接 PATCH 复活、价格计划校验、状态更新、懒过期、cache-only 持仓过滤、敏感信息脱敏、敏感 `trace_id` 拒绝和仅清理 `source_type=analysis` 历史绑定信号的历史删除联动。 +- [改进] #1390 P1 补充 Web decision-signals typed API wrapper 与契约隔离测试，暂不接入 UI。 +- [修复] #1390 收紧建议动作 legacy fallback：英文 `not to ...` 与 `avoid selling/reducing/trimming ...` 等否定/回避表达不再误判为买卖动作，Web 旧记录不再把中文金融上下文、`buy or sell`、多 guard 歧义文本或 `buyback` / `buy-back` / `buy back` / `selloff` / `sell-off` / `sell off` 等英文复合词渲染成 action badge，并在有结构化 `action` 时让回测/历史趋势等入口按界面语言显示 action 标签。 +- [改进] 完善运行时日志上下文，补充 logger name、触发来源、市场统计与实时行情预取链路状态，便于排查调度、API、Bot 和数据源降级路径。 +- [新功能] 新增分析任务与历史报告运行流快照 API，提供 lanes、nodes、edges、events、summary 等统一契约，并从任务队列、运行诊断和 AnalysisContextPack overview 构建脱敏数据流/信息流。 +- [新功能] Web 端为活跃任务、历史报告和大盘复盘报告补充运行流视图入口，支持查看运行摘要、拓扑节点、事件流和基础排障详情。 +- [修复] 修复历史报告运行流快照在混合时区事件时间戳下返回 500 的问题。 +- [改进] #1459 持仓管理页新增持仓账户删除入口，复用现有账户软删除接口，误建账户会从默认列表、快照、风险、录入入口和事件列表隐藏且不物理清理历史流水。 +- [修复] 修复运行流 live SSE 事件未复用快照层递归脱敏规则的问题，避免本地路径、prompt/raw response、代理头等敏感诊断字段在 refetch 前短暂暴露。 +- [修复] 修复 Web 首页分析任务卡片在窄侧栏下挤压股票信息、进度和运行诊断文案的问题。 +- [修复] 隔离个股分析自动生成的大盘上下文运行诊断，避免大盘复盘与个股报告共用 query_id 导致运行流重复展示“保存报告”和“推送通知”，并兼容通知跳过时 `attempts=0` 的运行流快照。 +- [改进] 运行流 active task 增加 provider 与 LLM started 实时事件，长耗时步骤开始时先显示 running 卡片，完成后复用同一节点更新结果，避免重复卡片。 +- [修复] 运行流为筹码分布补齐 provider started/result 事件，个股分析触发筹码数据源调用时可显示“筹码结构”运行卡片并记录降级尝试。 +- [修复] 修复个股运行流活跃任务后期 LLM/通知卡片临时重复、数据源聚合卡片过早显示成功，并为个股所属板块补齐运行流卡片。 - [修复] 发布说明生成查询 PR 作者失败时保留降级并输出包含 PR 编号和异常类型的 warning，便于排查 token、权限、网络或 GitHub API 异常。 diff --git a/docs/run-diagnostics-p3.md b/docs/run-diagnostics-p3.md index 32fb383b7..6d299fa5c 100644 --- a/docs/run-diagnostics-p3.md +++ b/docs/run-diagnostics-p3.md @@ -31,6 +31,7 @@ GET /api/v1/analysis/tasks/stream - 原有 task payload 字段保持不变。 - 当本次进度更新来自运行诊断时，可选追加 `flow_event` 字段；旧客户端忽略该字段即可。 - `flow_event` 使用与 `RunFlowSnapshot.events[]` 相同的脱敏事件结构：`id`、`timestamp`、`severity`、`type`、`node_id`、`title`、`message`、`metadata`。 +- active task 可追加 `provider_run_started` / `llm_run_started` 实时事件；这些事件只用于运行中的 running 卡片展示，完成后由同 `node_id` 的 `provider_run` / `llm_run` 结果事件覆盖状态，历史诊断仍以最终结果为准。 - 后端 TaskQueue 只为每个 active task 保留最近 N 条运行流事件，避免内存无限增长；完整历史仍以 `context_snapshot.diagnostics` 和历史 RunFlowSnapshot 为准。 示例： @@ -96,6 +97,9 @@ GET /api/v1/history/{record_id}/flow - `tasks/{task_id}/flow` 面向活跃任务。任务仍在内存队列中时优先返回当前任务快照；任务已完成时可按同一 `task_id/query_id` 尝试读取历史诊断。缺少诊断时返回 skeleton flow，不伪造 provider、LLM 或通知事件。 - `history/{record_id}/flow` 面向历史报告，支持历史记录主键 ID 或可解析的 `query_id`。普通个股分析与 `MARKET/market_review` 大盘复盘复用同一 `RunFlowSnapshot` 契约。 +- 同一页面触发个股分析时，个股流程可按需生成或复用当日大盘上下文；这不是独立的个股分析步骤，而是 Prompt 背景生成。后台会用独立 `market_context_*` query_id 与 `scope=daily_market_context` 保存该大盘上下文，避免与个股报告共用 query_id。 +- 为兼容早期已写入的混合诊断，运行流会在读取历史时按报告类型做低风险过滤：`MARKET/market_review` 记录隐藏个股行情、日线、技术、基本面与筹码 provider 节点；个股记录隐藏首次个股行情前的大盘新闻搜索，以及首次个股 LLM 前的大盘保存/通知节点。 +- 通知跳过或未配置时允许 `attempts=0`，运行流展示为 skipped，不再因 Pydantic 校验失败导致 `/flow` 返回 500。 - 快照顶层包含 `summary`、`lanes`、`nodes`、`edges`、`events` 和 `generated_at`。节点状态使用 `pending/running/success/failed/degraded/fallback/timeout/cancel_requested/cancelled/skipped/unknown`，其中用户取消类状态不会被映射成 `failed`。 - 旧历史、缺失 `context_snapshot.diagnostics` 或证据不足时，后端返回 `unknown` 或 skeleton 节点；Web 端按空/未知状态展示，不影响报告详情读取。 diff --git a/src/core/market_review.py b/src/core/market_review.py index f0b37871f..58ea4a52b 100644 --- a/src/core/market_review.py +++ b/src/core/market_review.py @@ -546,6 +546,11 @@ def _persist_market_review_history( diagnostic_snapshot = current_diagnostic_snapshot() if diagnostic_snapshot is not None: context_snapshot["diagnostics"] = diagnostic_snapshot + context_snapshot["analysis_context_pack_overview"] = _build_market_review_context_overview( + region=region, + report_language=report_language, + diagnostic_snapshot=diagnostic_snapshot, + ) db = DatabaseManager.get_instance() saved = db.save_analysis_history( @@ -556,11 +561,10 @@ def _persist_market_review_history( context_snapshot=context_snapshot, save_snapshot=True, ) - saved_history_id = ( - saved - if isinstance(saved, int) and not isinstance(saved, bool) and saved > 0 - else None - ) + saved_history_id = _resolve_saved_market_review_history_id( + db=db, + query_id=history_query_id, + ) if saved else None record_history_run( report_saved=bool(saved), metadata_saved=bool(saved), @@ -582,6 +586,93 @@ def _persist_market_review_history( return 0 +def _build_market_review_context_overview( + *, + region: str, + report_language: str, + diagnostic_snapshot: Optional[Dict[str, Any]], +) -> Dict[str, Any]: + """Build a low-sensitivity overview block for market-review run-flow rendering.""" + warnings: list[str] = [] + counts = { + "available": 1, + "missing": 0, + "not_supported": 0, + "fallback": 0, + "stale": 0, + "estimated": 0, + "partial": 0, + "fetch_failed": 0, + } + metadata: Dict[str, Any] = { + "trigger_source": "market_review", + "scope": "market_review", + "report_type": MARKET_REVIEW_REPORT_TYPE, + } + if isinstance(diagnostic_snapshot, dict): + metadata["trigger_source"] = diagnostic_snapshot.get("trigger_source") or metadata["trigger_source"] + metadata["scope"] = diagnostic_snapshot.get("scope") or metadata["scope"] + + label = "Market review" if report_language == "en" else "大盘复盘" + return { + "pack_version": "market_review/1.0", + "created_at": datetime.now().isoformat(), + "subject": { + "code": MARKET_REVIEW_HISTORY_CODE, + "stock_name": label, + "market": region, + }, + "blocks": [ + { + "key": MARKET_REVIEW_REPORT_TYPE, + "label": label, + "status": "available", + "source": MARKET_REVIEW_REPORT_TYPE, + "warnings": warnings, + "missing_reasons": [], + } + ], + "counts": counts, + "warnings": warnings, + "metadata": metadata, + "data_quality": { + "level": "good", + "overall_score": 100, + "available": 1, + "total": 1, + "missing": 0, + }, + } + + +def _resolve_saved_market_review_history_id( + *, + db: object, + query_id: str, +) -> Optional[int]: + """Resolve the real AnalysisHistory primary key after save_analysis_history returns row count.""" + try: + resolver = getattr(db, "get_latest_analysis_by_query_id", None) + if not callable(resolver): + return None + record = resolver( + query_id, + code=MARKET_REVIEW_HISTORY_CODE, + report_type=MARKET_REVIEW_REPORT_TYPE, + ) + record_id = getattr(record, "id", None) + return record_id if isinstance(record_id, int) and not isinstance(record_id, bool) else None + except TypeError: + try: + record = db.get_latest_analysis_by_query_id(query_id) + record_id = getattr(record, "id", None) + return record_id if isinstance(record_id, int) and not isinstance(record_id, bool) else None + except Exception: + return None + except Exception: + return None + + def _summarize_market_review(review_report: str, report_language: str) -> str: for line in (review_report or "").splitlines(): text = line.strip().lstrip("#").strip() diff --git a/src/core/pipeline.py b/src/core/pipeline.py index a9002f82c..8e102b7c3 100644 --- a/src/core/pipeline.py +++ b/src/core/pipeline.py @@ -65,6 +65,7 @@ from src.services.run_diagnostics import ( get_current_diagnostic_context, record_history_run, record_llm_run, + record_llm_run_started, record_notification_run, reset_run_diagnostic_context, sanitize_diagnostic_text, @@ -605,6 +606,10 @@ class StockAnalysisPipeline: self._emit_progress(64, f"{stock_name}：正在请求 LLM 生成报告") llm_started_at = time.monotonic() try: + record_llm_run_started( + model=getattr(self.config, "litellm_model", None), + call_type="analysis", + ) result = self.analyzer.analyze( enhanced_context, news_context=news_context, @@ -1127,6 +1132,10 @@ class StockAnalysisPipeline: message = f"请分析股票 {code} ({stock_name})，并生成决策仪表盘报告。" llm_started_at = time.monotonic() try: + record_llm_run_started( + model=getattr(self.config, "agent_litellm_model", None), + call_type="agent_analysis", + ) agent_result = executor.run(message, context=initial_context) except Exception as exc: record_llm_run( diff --git a/src/market_analyzer.py b/src/market_analyzer.py index ed2102032..e88cc3879 100644 --- a/src/market_analyzer.py +++ b/src/market_analyzer.py @@ -25,6 +25,7 @@ from src.search_service import SearchService from src.core.market_profile import get_profile, MarketProfile from src.core.market_strategy import get_market_strategy_blueprint from src.schemas.market_light import MarketLightSnapshot +from src.services.run_diagnostics import record_llm_run, record_llm_run_started from data_provider.base import DataFetcherManager logger = logging.getLogger(__name__) @@ -547,7 +548,35 @@ Focus on index trend, liquidity, and sector rotation to shape the next-session t logger.info("[大盘] %s action=generate_review status=start", self._log_context()) # Use the public generate_text() entry point - never access private analyzer attributes. - review = self.analyzer.generate_text(prompt, max_tokens=8192, temperature=0.7) + llm_started_at = time.perf_counter() + try: + record_llm_run_started( + provider="litellm", + model=getattr(self.config, "litellm_model", None), + call_type="market_review", + ) + review = self.analyzer.generate_text(prompt, max_tokens=8192, temperature=0.7) + except Exception as exc: + record_llm_run( + success=False, + provider="litellm", + model=getattr(self.config, "litellm_model", None), + call_type="market_review", + duration_ms=int((time.perf_counter() - llm_started_at) * 1000), + error_type=type(exc).__name__, + error_message=exc, + ) + raise + + record_llm_run( + success=bool(review), + provider="litellm", + model=getattr(self.config, "litellm_model", None), + call_type="market_review", + duration_ms=int((time.perf_counter() - llm_started_at) * 1000), + error_type=None if review else "EmptyResponse", + error_message=None if review else "empty market review response", + ) if review: logger.info( diff --git a/src/search_service.py b/src/search_service.py index cd083d133..bbb7a6700 100644 --- a/src/search_service.py +++ b/src/search_service.py @@ -38,7 +38,7 @@ from src.config import ( normalize_news_strategy_profile, resolve_news_window_days, ) -from src.services.run_diagnostics import record_provider_run +from src.services.run_diagnostics import record_provider_run, record_provider_run_started logger = logging.getLogger(__name__) @@ -3704,6 +3704,11 @@ class SearchService: started_at = time.monotonic() try: + record_provider_run_started( + data_type="news_search", + provider=provider.name, + operation="search_stock_news", + ) response = provider.search(query, provider_max_results, days=search_days, **search_kwargs) except Exception as exc: self._record_news_search_run( diff --git a/src/services/daily_market_context.py b/src/services/daily_market_context.py index 50b98b7c1..b502245ac 100644 --- a/src/services/daily_market_context.py +++ b/src/services/daily_market_context.py @@ -8,6 +8,7 @@ import logging import re import threading import time +import uuid from dataclasses import dataclass, field from datetime import date, datetime from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Tuple @@ -17,6 +18,10 @@ from src.core.market_review_lock import ( try_acquire_market_review_lock, ) from src.report_language import normalize_report_language +from src.services.run_diagnostics import ( + activate_run_diagnostic_context, + reset_run_diagnostic_context, +) from src.storage import DatabaseManager logger = logging.getLogger(__name__) @@ -413,23 +418,39 @@ class DailyMarketContextService: persist_market_review_history=persist_market_review_history, ) + caller_query_id = ( + current_query_id.strip() + if isinstance(current_query_id, str) and current_query_id.strip() + else None + ) + market_context_query_id = ( + f"market_context_{caller_query_id}_{region}" + if caller_query_id + else f"market_context_{uuid.uuid4().hex}_{region}" + ) + + diagnostic_token = None try: + diagnostic_token = activate_run_diagnostic_context( + trace_id=market_context_query_id, + query_id=market_context_query_id, + stock_code=MARKET_REVIEW_HISTORY_CODE, + trigger_source="daily_market_context", + scope="daily_market_context", + ) result = run_market_review( config=config, notifier=notifier, analyzer=analyzer, search_service=search_service, - query_id=( - current_query_id.strip() - if isinstance(current_query_id, str) and current_query_id.strip() - else None - ), + query_id=market_context_query_id, send_notification=False, merge_notification=False, override_region=region, return_structured=True, save_report_file=False, persist_history=persist_market_review_history, + trigger_source="daily_market_context", ) if ( @@ -451,11 +472,7 @@ class DailyMarketContextService: source="market_review_runtime", fallback_summary=fallback_summary, fallback_full_report=fallback_summary, - query_id=( - current_query_id.strip() - if isinstance(current_query_id, str) and current_query_id.strip() - else None - ), + query_id=caller_query_id, ) except Exception as exc: logger.warning( @@ -465,6 +482,8 @@ class DailyMarketContextService: ) return None finally: + if diagnostic_token is not None: + reset_run_diagnostic_context(diagnostic_token) if owns_lock: release_market_review_lock(lock_token) diff --git a/src/services/history_service.py b/src/services/history_service.py index 953818abc..834dd8152 100644 --- a/src/services/history_service.py +++ b/src/services/history_service.py @@ -284,7 +284,13 @@ class HistoryService: **market_fields, } - def _resolve_record(self, record_id: str): + def _resolve_record( + self, + record_id: str, + *, + code: Optional[str] = None, + report_type: Optional[str] = None, + ): """ Resolve a record_id parameter to an AnalysisHistory object. @@ -304,8 +310,15 @@ class HistoryService: return record except (ValueError, TypeError): pass - # Fall back to query_id lookup - return self.db.get_latest_analysis_by_query_id(record_id) + # Fall back to query_id lookup. Keep the old no-kwargs call for + # unfiltered paths so existing test doubles and integrations remain compatible. + if code is None and report_type is None: + return self.db.get_latest_analysis_by_query_id(record_id) + return self.db.get_latest_analysis_by_query_id( + record_id, + code=code, + report_type=report_type, + ) def resolve_and_get_detail(self, record_id: str) -> Optional[Dict[str, Any]]: """ @@ -373,14 +386,20 @@ class HistoryService: stock_code=getattr(record, "code", None), ) - def resolve_and_get_run_flow(self, record_id: str): + def resolve_and_get_run_flow( + self, + record_id: str, + *, + code: Optional[str] = None, + report_type: Optional[str] = None, + ): """ Resolve record_id and return a sanitized run-flow snapshot. Uses the same strict JSON parsing behavior as diagnostics so malformed persisted payloads surface as backend errors instead of partial graphs. """ - record = self._resolve_record(record_id) + record = self._resolve_record(record_id, code=code, report_type=report_type) if not record: return None diff --git a/src/services/run_diagnostics.py b/src/services/run_diagnostics.py index af3f10b33..b1c27223d 100644 --- a/src/services/run_diagnostics.py +++ b/src/services/run_diagnostics.py @@ -322,6 +322,7 @@ class RunDiagnosticContext: query_id: Optional[str] = None stock_code: Optional[str] = None trigger_source: Optional[str] = None + scope: Optional[str] = None provider_runs: List[ProviderRun] = field(default_factory=list) llm_runs: List[LLMRun] = field(default_factory=list) notification_runs: List[NotificationRun] = field(default_factory=list) @@ -329,17 +330,127 @@ class RunDiagnosticContext: event_sink: Optional[Callable[[Dict[str, Any]], None]] = None flow_event_index: int = 0 provider_attempt_index_by_type: Dict[str, int] = field(default_factory=dict) + provider_pending_attempt_index_by_key: Dict[str, List[int]] = field(default_factory=dict) + llm_attempt_index_by_type: Dict[str, int] = field(default_factory=dict) + llm_pending_attempt_index_by_key: Dict[str, List[int]] = field(default_factory=dict) + llm_pending_attempt_index_by_call_type: Dict[str, List[int]] = field(default_factory=dict) def record_provider_run(self, provider_run: ProviderRun) -> None: self.provider_runs.append(provider_run) data_type_key = _safe_event_key(provider_run.data_type) or "provider" + pending_key = _provider_pending_key( + provider_run.data_type, + provider_run.provider, + provider_run.operation, + ) + pending_indexes = self.provider_pending_attempt_index_by_key.get(pending_key) or [] + if pending_indexes: + attempt_index = pending_indexes.pop(0) + if pending_indexes: + self.provider_pending_attempt_index_by_key[pending_key] = pending_indexes + else: + self.provider_pending_attempt_index_by_key.pop(pending_key, None) + else: + attempt_index = self.provider_attempt_index_by_type.get(data_type_key, 0) + 1 + self.provider_attempt_index_by_type[data_type_key] = attempt_index + self._emit_flow_event(_provider_flow_event(self, provider_run, attempt_index)) + + def record_provider_run_started( + self, + *, + data_type: str, + provider: str, + operation: str, + ) -> None: + data_type_key = _safe_event_key(data_type) or "provider" attempt_index = self.provider_attempt_index_by_type.get(data_type_key, 0) + 1 self.provider_attempt_index_by_type[data_type_key] = attempt_index - self._emit_flow_event(_provider_flow_event(self, provider_run, attempt_index)) + pending_key = _provider_pending_key(data_type, provider, operation) + pending_indexes = self.provider_pending_attempt_index_by_key.get(pending_key) or [] + pending_indexes.append(attempt_index) + self.provider_pending_attempt_index_by_key[pending_key] = pending_indexes + self._emit_flow_event( + _provider_started_flow_event( + self, + data_type=data_type, + provider=provider, + operation=operation, + index=attempt_index, + ) + ) def record_llm_run(self, llm_run: LLMRun) -> None: self.llm_runs.append(llm_run) - self._emit_flow_event(_llm_flow_event(self, llm_run, len(self.llm_runs))) + call_type_key = _safe_event_key(llm_run.call_type) or "analysis" + pending_key = _llm_pending_key(llm_run.call_type, llm_run.provider, llm_run.model) + pending_indexes = self.llm_pending_attempt_index_by_key.get(pending_key) or [] + if pending_indexes: + attempt_index = pending_indexes.pop(0) + if pending_indexes: + self.llm_pending_attempt_index_by_key[pending_key] = pending_indexes + else: + self.llm_pending_attempt_index_by_key.pop(pending_key, None) + self._remove_llm_pending_call_type_index(call_type_key, attempt_index) + else: + call_type_pending_indexes = self.llm_pending_attempt_index_by_call_type.get(call_type_key) or [] + if call_type_pending_indexes: + attempt_index = call_type_pending_indexes.pop(0) + if call_type_pending_indexes: + self.llm_pending_attempt_index_by_call_type[call_type_key] = call_type_pending_indexes + else: + self.llm_pending_attempt_index_by_call_type.pop(call_type_key, None) + self._remove_llm_pending_exact_index(attempt_index) + else: + attempt_index = self.llm_attempt_index_by_type.get(call_type_key, 0) + 1 + self.llm_attempt_index_by_type[call_type_key] = attempt_index + self._emit_flow_event(_llm_flow_event(self, llm_run, attempt_index)) + + def _remove_llm_pending_call_type_index(self, call_type_key: str, attempt_index: int) -> None: + pending_indexes = self.llm_pending_attempt_index_by_call_type.get(call_type_key) or [] + if attempt_index not in pending_indexes: + return + pending_indexes = [index for index in pending_indexes if index != attempt_index] + if pending_indexes: + self.llm_pending_attempt_index_by_call_type[call_type_key] = pending_indexes + else: + self.llm_pending_attempt_index_by_call_type.pop(call_type_key, None) + + def _remove_llm_pending_exact_index(self, attempt_index: int) -> None: + for pending_key, pending_indexes in list(self.llm_pending_attempt_index_by_key.items()): + if attempt_index not in pending_indexes: + continue + pending_indexes = [index for index in pending_indexes if index != attempt_index] + if pending_indexes: + self.llm_pending_attempt_index_by_key[pending_key] = pending_indexes + else: + self.llm_pending_attempt_index_by_key.pop(pending_key, None) + + def record_llm_run_started( + self, + *, + call_type: str = "analysis", + provider: Optional[str] = None, + model: Optional[str] = None, + ) -> None: + call_type_key = _safe_event_key(call_type) or "analysis" + attempt_index = self.llm_attempt_index_by_type.get(call_type_key, 0) + 1 + self.llm_attempt_index_by_type[call_type_key] = attempt_index + pending_key = _llm_pending_key(call_type, provider, model) + pending_indexes = self.llm_pending_attempt_index_by_key.get(pending_key) or [] + pending_indexes.append(attempt_index) + self.llm_pending_attempt_index_by_key[pending_key] = pending_indexes + call_type_pending_indexes = self.llm_pending_attempt_index_by_call_type.get(call_type_key) or [] + call_type_pending_indexes.append(attempt_index) + self.llm_pending_attempt_index_by_call_type[call_type_key] = call_type_pending_indexes + self._emit_flow_event( + _llm_started_flow_event( + self, + call_type=call_type, + provider=provider, + model=model, + index=attempt_index, + ) + ) def record_notification_run(self, notification_run: NotificationRun) -> None: self.notification_runs.append(notification_run) @@ -368,6 +479,7 @@ class RunDiagnosticContext: "query_id": self.query_id, "stock_code": self.stock_code, "trigger_source": self.trigger_source, + "scope": self.scope, "provider_runs": [run.to_dict() for run in self.provider_runs], "llm_runs": [run.to_dict() for run in self.llm_runs], "notification_runs": [run.to_dict() for run in self.notification_runs], @@ -386,6 +498,7 @@ def activate_run_diagnostic_context( query_id: Optional[str] = None, stock_code: Optional[str] = None, trigger_source: Optional[str] = None, + scope: Optional[str] = None, event_sink: Optional[Callable[[Dict[str, Any]], None]] = None, ) -> Token: """Activate a diagnostic context and return its reset token.""" @@ -395,6 +508,7 @@ def activate_run_diagnostic_context( query_id=query_id, stock_code=stock_code, trigger_source=trigger_source, + scope=scope, event_sink=event_sink, ) return _CURRENT_CONTEXT.set(context) @@ -429,6 +543,7 @@ _DATA_TYPE_LABELS = { "news_search": "新闻舆情", "fundamental": "基本面", "fundamentals": "基本面", + "belong_boards": "所属板块", "chip": "筹码结构", } @@ -445,6 +560,21 @@ def _clean_metadata(value: Dict[str, Any]) -> Dict[str, Any]: } +def _provider_pending_key(data_type: Any, provider: Any, operation: Any) -> str: + return "|".join( + ( + _safe_event_key(data_type) or "provider", + _safe_event_key(provider) or "unknown", + _safe_event_key(operation) or "operation", + ) + ) + + +def _llm_pending_key(call_type: Any, provider: Any, model: Any) -> str: + _ = (provider, model) + return _safe_event_key(call_type) or "analysis" + + def _flow_status_for_success(success: bool, *, fallback: bool = False, skipped: bool = False) -> str: if skipped: return "skipped" @@ -469,6 +599,49 @@ def _started_at_from_end_and_duration(end: Any, duration_ms: Optional[int]) -> O return (parsed - timedelta(milliseconds=duration_ms)).isoformat() +def _provider_started_flow_event( + context: RunDiagnosticContext, + *, + data_type: str, + provider: str, + operation: str, + index: int, +) -> Dict[str, Any]: + data_type_key = _safe_event_key(data_type) or "provider" + provider_key = _safe_event_key(provider) or "unknown" + label = _DATA_TYPE_LABELS.get(data_type_key, data_type_key) + node_id = f"provider_{data_type_key}_{provider_key}_{index}" + timestamp = datetime.now().isoformat() + message = f"{label} {provider} 调用中" + return { + "timestamp": timestamp, + "severity": "info", + "type": "provider_run_started", + "node_id": node_id, + "title": f"{label}开始", + "message": sanitize_diagnostic_text(message, max_length=220), + "metadata": _clean_metadata( + { + "trace_id": context.trace_id, + "provider": provider, + "data_type": data_type, + "operation": operation, + "node": { + "id": node_id, + "lane": "data_source", + "kind": "data_source", + "label": f"{label} · {provider}", + "status": "running", + "provider": provider, + "started_at": timestamp, + "attempts": 1, + "message": message, + }, + } + ), + } + + def _provider_flow_event( context: RunDiagnosticContext, run: ProviderRun, @@ -522,6 +695,48 @@ def _provider_flow_event( } +def _llm_started_flow_event( + context: RunDiagnosticContext, + *, + call_type: str, + provider: Optional[str], + model: Optional[str], + index: int, +) -> Dict[str, Any]: + call_type_key = _safe_event_key(call_type) or "analysis" + display_model = model or provider or "unknown" + node_id = f"llm_{call_type_key}_{index}" + timestamp = datetime.now().isoformat() + message = f"LLM {display_model} 调用中" + return { + "timestamp": timestamp, + "severity": "info", + "type": "llm_run_started", + "node_id": node_id, + "title": "LLM 开始", + "message": sanitize_diagnostic_text(message, max_length=220), + "metadata": _clean_metadata( + { + "trace_id": context.trace_id, + "provider": provider, + "model": model, + "call_type": call_type, + "node": { + "id": node_id, + "lane": "analysis", + "kind": "model", + "label": "LLM 生成", + "status": "running", + "provider": display_model, + "started_at": timestamp, + "attempts": 1, + "message": message, + }, + } + ), + } + + def _llm_flow_event( context: RunDiagnosticContext, run: LLMRun, @@ -692,6 +907,27 @@ def record_provider_run( logger.warning("provider diagnostic record failed: %s", exc) +def record_provider_run_started( + *, + data_type: str, + provider: str, + operation: str, +) -> None: + """Emit a live provider-start event without changing persisted diagnostics.""" + context = get_current_diagnostic_context() + if context is None: + return + + try: + context.record_provider_run_started( + data_type=data_type, + provider=provider, + operation=operation, + ) + except Exception as exc: # pragma: no cover - defensive fail-open guard + logger.warning("provider started diagnostic record failed: %s", exc) + + def record_llm_run( *, success: bool, @@ -728,6 +964,27 @@ def record_llm_run( logger.warning("llm diagnostic record failed: %s", exc) +def record_llm_run_started( + *, + provider: Optional[str] = None, + model: Optional[str] = None, + call_type: str = "analysis", +) -> None: + """Emit a live LLM-start event without changing persisted diagnostics.""" + context = get_current_diagnostic_context() + if context is None: + return + + try: + context.record_llm_run_started( + provider=provider, + model=model, + call_type=call_type, + ) + except Exception as exc: # pragma: no cover - defensive fail-open guard + logger.warning("llm started diagnostic record failed: %s", exc) + + def record_notification_run( *, channel: str, diff --git a/src/services/run_flow.py b/src/services/run_flow.py index 6ac2e2057..93b05ea4c 100644 --- a/src/services/run_flow.py +++ b/src/services/run_flow.py @@ -46,6 +46,7 @@ _DATA_TYPE_LABELS = { "news_search": "新闻舆情", "fundamental": "基本面", "fundamentals": "基本面", + "belong_boards": "所属板块", "chip": "筹码结构", } @@ -58,6 +59,7 @@ _DATA_TYPE_TO_BLOCK_KEY = { "news_search": "news", "fundamental": "fundamentals", "fundamentals": "fundamentals", + "belong_boards": "fundamentals", "chip": "chip", } @@ -153,6 +155,7 @@ def build_task_run_flow_snapshot( _as_list(getattr(task, "flow_events", None)), flow_status=flow_status, ) + _prune_active_skeleton_tail(nodes, edges) summary = _build_summary( nodes, @@ -188,6 +191,7 @@ def build_history_run_flow_snapshot( snapshot = _as_mapping(context_snapshot if context_snapshot is not None else getattr(record, "context_snapshot", None)) raw = _as_mapping(raw_result if raw_result is not None else getattr(record, "raw_result", None)) diagnostics = _as_mapping(snapshot.get("diagnostics")) if snapshot else {} + diagnostics = _normalize_history_diagnostics_for_record(record, snapshot, diagnostics) overview = extract_analysis_context_pack_overview(snapshot) if snapshot else None overview_metadata = overview.get("metadata") if isinstance((overview or {}).get("metadata"), Mapping) else {} @@ -694,7 +698,7 @@ def _append_notification_runs( status=status, provider=channel, ended_at=timestamp, - attempts=_safe_int(run.get("attempts")) or 1, + attempts=_safe_int(run.get("attempts")) if _safe_int(run.get("attempts")) is not None else 1, message=message, metadata={ "channel": channel, @@ -721,6 +725,91 @@ def _append_notification_runs( return count +_STOCK_CONTEXT_PROVIDER_DATA_TYPES = { + "realtime_quote", + "daily_data", + "daily_bars", + "technical", + "fundamental", + "fundamentals", + "belong_boards", + "chip", +} + + +def _normalize_history_diagnostics_for_record( + record: Any, + snapshot: Dict[str, Any], + diagnostics: Dict[str, Any], +) -> Dict[str, Any]: + if not diagnostics: + return diagnostics + + normalized = dict(diagnostics) + report_type = _safe_key(getattr(record, "report_type", None)) + code = _safe_text(getattr(record, "code", None), max_length=32) + report_kind = _safe_key(snapshot.get("report_kind")) if snapshot else "" + + if report_type == "market_review" or report_kind == "market_review" or (code or "").upper() == "MARKET": + normalized["stock_code"] = "MARKET" + normalized.setdefault("scope", "market_review") + normalized["provider_runs"] = [ + run + for run in _as_list(normalized.get("provider_runs")) + if _safe_key(_as_mapping(run).get("data_type")) not in _STOCK_CONTEXT_PROVIDER_DATA_TYPES + ] + return normalized + + first_llm_at = _first_timestamp(_as_list(normalized.get("llm_runs"))) + if first_llm_at is not None: + normalized["history_runs"] = [ + run + for run in _as_list(normalized.get("history_runs")) + if not _timestamp_before(_as_mapping(run).get("created_at"), first_llm_at) + ] + normalized["notification_runs"] = [ + run + for run in _as_list(normalized.get("notification_runs")) + if not _timestamp_before(_as_mapping(run).get("created_at"), first_llm_at) + ] + + first_stock_data_at = _first_timestamp( + [ + run + for run in _as_list(normalized.get("provider_runs")) + if _safe_key(_as_mapping(run).get("data_type")) in _STOCK_CONTEXT_PROVIDER_DATA_TYPES + ] + ) + if first_stock_data_at is not None: + normalized["provider_runs"] = [ + run + for run in _as_list(normalized.get("provider_runs")) + if _safe_key(_as_mapping(run).get("data_type")) != "news_search" + or not _timestamp_before(_as_mapping(run).get("created_at"), first_stock_data_at) + ] + return normalized + + +def _first_timestamp(items: List[Any]) -> Optional[datetime]: + timestamps = [ + parsed + for parsed in (_datetime_for_elapsed(_as_mapping(item).get("created_at")) for item in items) + if parsed is not None + ] + return min(timestamps) if timestamps else None + + +def _timestamp_before(value: Any, boundary: datetime) -> bool: + parsed = _datetime_for_elapsed(value) + if parsed is None: + return False + if parsed.tzinfo is None and boundary.tzinfo is not None: + parsed = parsed.replace(tzinfo=boundary.tzinfo) + elif parsed.tzinfo is not None and boundary.tzinfo is None: + boundary = boundary.replace(tzinfo=parsed.tzinfo) + return parsed < boundary + + def _put_skeleton_tail( nodes: Dict[str, Dict[str, Any]], edges: List[Dict[str, Any]], @@ -771,6 +860,26 @@ def _put_skeleton_tail( _append_edge(edges, "history_save", "notification", "control", downstream_status, label="通知") +def _prune_active_skeleton_tail( + nodes: Dict[str, Dict[str, Any]], + edges: List[Dict[str, Any]], +) -> None: + remove_node_ids = set() + if "llm" in nodes and any(node_id.startswith("llm_") for node_id in nodes): + remove_node_ids.add("llm") + if "notification" in nodes and any(node_id.startswith("notification_") for node_id in nodes): + remove_node_ids.add("notification") + if not remove_node_ids: + return + for node_id in remove_node_ids: + nodes.pop(node_id, None) + edges[:] = [ + edge + for edge in edges + if edge.get("from") not in remove_node_ids and edge.get("to") not in remove_node_ids + ] + + def _append_task_events(events: List[Dict[str, Any]], task: Any, flow_status: str) -> None: _append_event( events, @@ -873,15 +982,28 @@ def _append_active_flow_events( ) event_type = _safe_key(event.get("type")) or "event" + provider_data_type = None + provider_run = None + if event_type in {"provider_run", "provider_run_started"} and node_id and node_id in nodes: + provider_data_type = _safe_key(metadata.get("data_type") or "provider") + provider_run = { + "provider": metadata.get("provider") or nodes[node_id].get("provider"), + "success": event.get("severity") == "success" or nodes[node_id].get("status") in {"success", "fallback"}, + "fallback_from": metadata.get("fallback_from"), + "fallback_to": metadata.get("fallback_to"), + } + + if node_id and node_id in nodes and node_id in known_node_ids: + _refresh_incoming_edge_status(edges, node_id, nodes[node_id].get("status")) + if provider_data_type and provider_run: + last_provider_node_by_type[provider_data_type] = (node_id, provider_run) + elif event_type in {"llm_run", "llm_run_started"}: + last_llm_node = node_id + elif event_type == "history_run": + last_history_node = node_id + if node_id and node_id in nodes and node_id not in known_node_ids: - if event_type == "provider_run": - provider_data_type = _safe_key(metadata.get("data_type") or "provider") - provider_run = { - "provider": metadata.get("provider") or nodes[node_id].get("provider"), - "success": event.get("severity") == "success" or nodes[node_id].get("status") in {"success", "fallback"}, - "fallback_from": metadata.get("fallback_from"), - "fallback_to": metadata.get("fallback_to"), - } + if provider_data_type and provider_run: previous_provider = last_provider_node_by_type.get(provider_data_type) if previous_provider: previous_provider_node, previous_provider_run = previous_provider @@ -897,7 +1019,7 @@ def _append_active_flow_events( else: _append_edge(edges, "task_queue", node_id, "control", nodes[node_id].get("status", "unknown"), label="调用") last_provider_node_by_type[provider_data_type] = (node_id, provider_run) - elif event_type == "llm_run": + elif event_type in {"llm_run", "llm_run_started"}: anchor = "analysis_pipeline" if "analysis_pipeline" in nodes else "task_queue" _append_edge(edges, anchor, node_id, "data", nodes[node_id].get("status", "unknown"), label="生成") last_llm_node = node_id @@ -1167,7 +1289,19 @@ def _append_edge( metadata: Optional[Any] = None, ) -> None: edge_id = f"{from_node}_to_{to_node}_{kind}" - if any(edge["id"] == edge_id for edge in edges): + for edge in edges: + if edge["id"] != edge_id: + continue + edge["status"] = _valid_status(status) + safe_label = _safe_text(label, max_length=40) + if safe_label: + edge["label"] = safe_label + safe_message = _safe_text(message, max_length=180) + if safe_message: + edge["message"] = safe_message + safe_metadata = _sanitize_metadata(metadata or {}) + if safe_metadata: + edge["metadata"] = safe_metadata return edges.append( { @@ -1183,6 +1317,19 @@ def _append_edge( ) +def _refresh_incoming_edge_status( + edges: List[Dict[str, Any]], + node_id: Optional[str], + status: Optional[Any], +) -> None: + if not node_id or status is None: + return + valid_status = _valid_status(status) + for edge in edges: + if edge.get("to") == node_id: + edge["status"] = valid_status + + def _append_event( events: List[Dict[str, Any]], event_type: str, diff --git a/src/storage.py b/src/storage.py index 85515ed4c..5d7542186 100644 --- a/src/storage.py +++ b/src/storage.py @@ -1811,7 +1811,13 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta): ) return list(results) - def get_latest_analysis_by_query_id(self, query_id: str) -> Optional[AnalysisHistory]: + def get_latest_analysis_by_query_id( + self, + query_id: str, + *, + code: Optional[str] = None, + report_type: Optional[str] = None, + ) -> Optional[AnalysisHistory]: """ 根据 query_id 查询最新一条分析历史记录 @@ -1819,14 +1825,22 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta): Args: query_id: 分析记录关联的 query_id + code: 可选股票代码过滤，用于区分同一 query_id 下的 MARKET 与个股记录 + report_type: 可选报告类型过滤 Returns: AnalysisHistory 对象，不存在返回 None """ with self.get_session() as session: + conditions = [AnalysisHistory.query_id == query_id] + if code: + conditions.append(AnalysisHistory.code == code) + if report_type: + conditions.append(AnalysisHistory.report_type == report_type) + result = session.execute( select(AnalysisHistory) - .where(AnalysisHistory.query_id == query_id) + .where(and_(*conditions)) .order_by(desc(AnalysisHistory.created_at)) .limit(1) ).scalars().first() diff --git a/tests/test_belong_boards_run_flow.py b/tests/test_belong_boards_run_flow.py new file mode 100644 index 000000000..91e6e2773 --- /dev/null +++ b/tests/test_belong_boards_run_flow.py @@ -0,0 +1,116 @@ +# -*- coding: utf-8 -*- +"""Regression tests for belong-board run-flow diagnostics.""" + +from src.services.run_diagnostics import ( + activate_run_diagnostic_context, + current_diagnostic_snapshot, + reset_run_diagnostic_context, +) +from data_provider.base import DataFetcherManager + + +class _BoardFetcher: + def __init__(self, name: str, result): + self.name = name + self.priority = 0 + self._result = result + self.calls = 0 + + def get_belong_board(self, _stock_code: str): + self.calls += 1 + return self._result + + +class _FailingBoardFetcher(_BoardFetcher): + def __init__(self, name: str, error: Exception): + super().__init__(name, []) + self._error = error + + def get_belong_board(self, _stock_code: str): + self.calls += 1 + raise self._error + + +def _capture_belong_board_run(manager: DataFetcherManager): + flow_events = [] + token = activate_run_diagnostic_context( + trace_id="trace-boards", + task_id="task-boards", + query_id="query-boards", + stock_code="600519", + trigger_source="api", + event_sink=flow_events.append, + ) + try: + boards = manager.get_belong_boards("600519") + diagnostics = current_diagnostic_snapshot() + finally: + reset_run_diagnostic_context(token) + return boards, diagnostics, flow_events + + +def test_get_belong_boards_records_successful_provider_run(): + manager = DataFetcherManager( + fetchers=[ + _BoardFetcher( + "BoardFetcher", + [{"name": "白酒", "type": "行业"}], + ) + ] + ) + + boards, diagnostics, flow_events = _capture_belong_board_run(manager) + + assert boards + assert diagnostics is not None + provider_runs = diagnostics["provider_runs"] + assert len(provider_runs) == 1 + assert provider_runs[0]["data_type"] == "belong_boards" + assert provider_runs[0]["provider"] == "BoardFetcher" + assert provider_runs[0]["operation"] == "get_belong_board" + assert provider_runs[0]["success"] is True + assert provider_runs[0]["record_count"] == len(boards) + assert [event["type"] for event in flow_events] == ["provider_run_started", "provider_run"] + assert flow_events[0]["node_id"] == flow_events[1]["node_id"] + assert flow_events[0]["node_id"] == "provider_belong_boards_boardfetcher_1" + + +def test_get_belong_boards_records_empty_attempt_and_fallback(): + manager = DataFetcherManager( + fetchers=[ + _BoardFetcher("EmptyBoardFetcher", []), + _BoardFetcher("FallbackBoardFetcher", [{"name": "电力设备", "type": "行业"}]), + ] + ) + + boards, diagnostics, flow_events = _capture_belong_board_run(manager) + + assert boards + assert diagnostics is not None + provider_runs = diagnostics["provider_runs"] + assert [run["provider"] for run in provider_runs] == ["EmptyBoardFetcher", "FallbackBoardFetcher"] + assert [run["success"] for run in provider_runs] == [False, True] + assert provider_runs[0]["error_type"] == "empty" + assert provider_runs[0]["fallback_to"] == "FallbackBoardFetcher" + assert len(flow_events) == 4 + + +def test_get_belong_boards_records_exception_attempt_and_fallback(): + manager = DataFetcherManager( + fetchers=[ + _FailingBoardFetcher("FailingBoardFetcher", RuntimeError("board source down")), + _BoardFetcher("FallbackBoardFetcher", [{"name": "电力设备", "type": "行业"}]), + ] + ) + + boards, diagnostics, flow_events = _capture_belong_board_run(manager) + + assert boards + assert diagnostics is not None + provider_runs = diagnostics["provider_runs"] + assert [run["provider"] for run in provider_runs] == ["FailingBoardFetcher", "FallbackBoardFetcher"] + assert provider_runs[0]["success"] is False + assert provider_runs[0]["error_type"] == "RuntimeError" + assert provider_runs[0]["fallback_to"] == "FallbackBoardFetcher" + assert provider_runs[1]["success"] is True + assert len(flow_events) == 4 diff --git a/tests/test_chip_distribution_manager.py b/tests/test_chip_distribution_manager.py index 38452cd94..005b8ee1c 100644 --- a/tests/test_chip_distribution_manager.py +++ b/tests/test_chip_distribution_manager.py @@ -20,6 +20,41 @@ class _ChipFetcher: return self._result +class _FailingChipFetcher(_ChipFetcher): + def __init__(self, name: str, priority: int, error: Exception): + super().__init__(name, priority, None) + self._error = error + + def get_chip_distribution(self, stock_code: str): + self.calls += 1 + raise self._error + + +def _run_with_chip_diagnostics(manager: DataFetcherManager): + from src.services.run_diagnostics import ( + activate_run_diagnostic_context, + current_diagnostic_snapshot, + reset_run_diagnostic_context, + ) + + flow_events = [] + token = activate_run_diagnostic_context( + trace_id="trace-chip", + task_id="task-chip", + query_id="query-chip", + stock_code="600519", + trigger_source="api", + event_sink=flow_events.append, + ) + try: + with patch("src.config.get_config", return_value=SimpleNamespace(enable_chip_distribution=True)): + chip = manager.get_chip_distribution("600519") + diagnostics = current_diagnostic_snapshot() + finally: + reset_run_diagnostic_context(token) + return chip, diagnostics, flow_events + + def test_manager_skips_placeholder_chip_distribution_and_tries_next_fetcher(): get_chip_circuit_breaker().reset() empty_chip = ChipDistribution(code="600519") @@ -36,10 +71,26 @@ def test_manager_skips_placeholder_chip_distribution_and_tries_next_fetcher(): ] ) - with patch("src.config.get_config", return_value=SimpleNamespace(enable_chip_distribution=True)): - chip = manager.get_chip_distribution("600519") + chip, diagnostics, flow_events = _run_with_chip_diagnostics(manager) assert chip is valid_chip + assert diagnostics is not None + provider_runs = diagnostics["provider_runs"] + assert [run["data_type"] for run in provider_runs] == ["chip", "chip"] + assert [run["success"] for run in provider_runs] == [False, True] + assert provider_runs[0]["fallback_to"] == "ValidFetcher" + assert provider_runs[0]["record_count"] == 0 + assert provider_runs[1]["record_count"] == 1 + assert [event["type"] for event in flow_events] == [ + "provider_run_started", + "provider_run", + "provider_run_started", + "provider_run", + ] + assert flow_events[0]["node_id"] == flow_events[1]["node_id"] + assert flow_events[2]["node_id"] == flow_events[3]["node_id"] + assert flow_events[0]["node_id"] == "provider_chip_emptyfetcher_1" + assert flow_events[2]["node_id"] == "provider_chip_validfetcher_2" def test_manager_accepts_zero_concentration_chip_distribution(): @@ -61,9 +112,46 @@ def test_manager_accepts_zero_concentration_chip_distribution(): fallback_fetcher = _ChipFetcher("FallbackFetcher", 1, fallback_chip) manager = DataFetcherManager(fetchers=[zero_fetcher, fallback_fetcher]) - with patch("src.config.get_config", return_value=SimpleNamespace(enable_chip_distribution=True)): - chip = manager.get_chip_distribution("600519") + chip, diagnostics, flow_events = _run_with_chip_diagnostics(manager) assert chip is zero_concentration_chip assert zero_fetcher.calls == 1 assert fallback_fetcher.calls == 0 + assert diagnostics is not None + assert len(diagnostics["provider_runs"]) == 1 + assert diagnostics["provider_runs"][0]["data_type"] == "chip" + assert diagnostics["provider_runs"][0]["success"] is True + assert [event["type"] for event in flow_events] == ["provider_run_started", "provider_run"] + assert flow_events[0]["node_id"] == flow_events[1]["node_id"] + + +def test_manager_records_failed_chip_attempt_and_falls_back_to_next_fetcher(): + get_chip_circuit_breaker().reset() + valid_chip = ChipDistribution( + code="600519", + profit_ratio=0.61, + avg_cost=12.3, + concentration_90=0.13, + ) + failing_fetcher = _FailingChipFetcher("FailingFetcher", 0, RuntimeError("temporary chip failure")) + fallback_fetcher = _ChipFetcher("FallbackFetcher", 1, valid_chip) + manager = DataFetcherManager(fetchers=[failing_fetcher, fallback_fetcher]) + + chip, diagnostics, flow_events = _run_with_chip_diagnostics(manager) + + assert chip is valid_chip + assert failing_fetcher.calls == 1 + assert fallback_fetcher.calls == 1 + assert diagnostics is not None + provider_runs = diagnostics["provider_runs"] + assert [run["provider"] for run in provider_runs] == ["FailingFetcher", "FallbackFetcher"] + assert provider_runs[0]["success"] is False + assert provider_runs[0]["error_type"] == "RuntimeError" + assert provider_runs[0]["fallback_to"] == "FallbackFetcher" + assert provider_runs[1]["success"] is True + assert [event["type"] for event in flow_events] == [ + "provider_run_started", + "provider_run", + "provider_run_started", + "provider_run", + ] diff --git a/tests/test_daily_market_context.py b/tests/test_daily_market_context.py index 695fec48d..7409fbdfd 100644 --- a/tests/test_daily_market_context.py +++ b/tests/test_daily_market_context.py @@ -463,7 +463,7 @@ def test_reuses_same_run_history_when_saved_under_different_wall_clock_date() -> run_review.assert_not_called() -def test_get_context_passes_current_query_id_to_market_review_when_generating() -> None: +def test_get_context_uses_isolated_market_context_query_id_when_generating() -> None: db = MagicMock() db.get_analysis_history.return_value = [] service = DailyMarketContextService( @@ -503,8 +503,10 @@ def test_get_context_passes_current_query_id_to_market_review_when_generating() assert context is not None assert context.source == "market_review_runtime" + assert context.query_id == "query-1381" run_review.assert_called_once() - assert run_review.call_args.kwargs["query_id"] == "query-1381" + assert run_review.call_args.kwargs["query_id"] == "market_context_query-1381_cn" + assert run_review.call_args.kwargs["trigger_source"] == "daily_market_context" acquire_lock.assert_called_once() release_lock.assert_called_once_with(lock_token) diff --git a/tests/test_market_review.py b/tests/test_market_review.py index 09db29633..5b13cfe66 100644 --- a/tests/test_market_review.py +++ b/tests/test_market_review.py @@ -431,6 +431,13 @@ class MarketReviewLocalizationTestCase(unittest.TestCase): ) self.assertEqual(saved, 1) + with DatabaseManager.get_instance().get_session() as session: + row = session.query(AnalysisHistory).filter( + AnalysisHistory.query_id == "market-task-001" + ).first() + self.assertIsNotNone(row) + snapshot = json.loads(row.context_snapshot or "{}") + self.assertIn("analysis_context_pack_overview", snapshot) db = DatabaseManager.get_instance() with db.get_session() as session: row = session.query(AnalysisHistory).filter( diff --git a/tests/test_run_flow.py b/tests/test_run_flow.py index f4fec5ecd..e99dfeb08 100644 --- a/tests/test_run_flow.py +++ b/tests/test_run_flow.py @@ -24,7 +24,11 @@ from src.services.run_flow import ( from src.services.run_diagnostics import ( activate_run_diagnostic_context, current_diagnostic_snapshot, + record_llm_run, + record_llm_run_started, + record_notification_run, record_provider_run, + record_provider_run_started, reset_run_diagnostic_context, ) from src.services.task_queue import AnalysisTaskQueue, TaskInfo, TaskStatus @@ -191,7 +195,13 @@ class _FakeHistoryDb: def get_analysis_history_by_id(self, record_id: int): return self.record if self.record is not None and record_id == self.record.id else None - def get_latest_analysis_by_query_id(self, query_id: str): + def get_latest_analysis_by_query_id(self, query_id: str, *, code: str | None = None, report_type: str | None = None): + if self.record is None or query_id != self.record.query_id: + return None + if code is not None and self.record.code != code: + return None + if report_type is not None and self.record.report_type != report_type: + return None return self.record if self.record is not None and query_id == self.record.query_id else None @@ -205,6 +215,10 @@ class _FakeMarketReviewDb: self.saved_context_snapshot = kwargs.get("context_snapshot") return self.save_result + def get_latest_analysis_by_query_id(self, query_id: str, *, code: str | None = None, report_type: str | None = None): + _ = (query_id, code, report_type) + return SimpleNamespace(id=42) + def update_analysis_history_diagnostics(self, *, query_id: str, code: str, diagnostics: dict) -> None: _ = (query_id, code) self.updated_diagnostics = diagnostics @@ -416,6 +430,213 @@ class RunFlowTestCase(unittest.TestCase): f"{node_id}.{field}", ) + def test_active_started_events_update_same_provider_and_llm_nodes(self) -> None: + flow_events: list[dict] = [] + token = activate_run_diagnostic_context( + trace_id="trace-started", + task_id="task-started", + query_id="query-started", + stock_code="600519", + trigger_source="api", + event_sink=flow_events.append, + ) + try: + record_provider_run_started( + data_type="daily_data", + provider="DailyFetcher", + operation="get_daily_data", + ) + record_provider_run( + data_type="daily_data", + provider="DailyFetcher", + operation="get_daily_data", + success=True, + latency_ms=120, + record_count=30, + ) + record_llm_run_started( + model="deepseek-chat", + call_type="analysis", + ) + record_llm_run( + success=True, + model="deepseek-chat", + call_type="analysis", + duration_ms=900, + ) + finally: + reset_run_diagnostic_context(token) + + snapshot = build_task_run_flow_snapshot( + TaskInfo( + task_id="task-started", + trace_id="trace-started", + stock_code="600519", + stock_name="贵州茅台", + status=TaskStatus.PROCESSING, + created_at=datetime(2026, 6, 8, 10, 0, 0), + flow_events=flow_events, + ) + ) + + provider_nodes = [node for node in snapshot.nodes if node.id == "provider_daily_data_dailyfetcher_1"] + llm_nodes = [node for node in snapshot.nodes if node.id == "llm_analysis_1"] + provider_edges = [ + edge for edge in snapshot.edges + if edge.to_node == "provider_daily_data_dailyfetcher_1" + ] + llm_edges = [ + edge for edge in snapshot.edges + if edge.to_node == "llm_analysis_1" + ] + + self.assertEqual(len(provider_nodes), 1) + self.assertEqual(provider_nodes[0].status, "success") + self.assertEqual(provider_nodes[0].record_count, 30) + self.assertTrue(provider_edges) + self.assertTrue(all(edge.status == "success" for edge in provider_edges)) + self.assertEqual(len(llm_nodes), 1) + self.assertEqual(llm_nodes[0].status, "success") + self.assertTrue(llm_edges) + self.assertTrue(all(edge.status == "success" for edge in llm_edges)) + self.assertIn("provider_run_started", {event.type for event in snapshot.events}) + self.assertIn("llm_run_started", {event.type for event in snapshot.events}) + + def test_active_chip_started_event_updates_same_provider_node(self) -> None: + flow_events: list[dict] = [] + token = activate_run_diagnostic_context( + trace_id="trace-chip-started", + task_id="task-chip-started", + query_id="query-chip-started", + stock_code="600519", + trigger_source="api", + event_sink=flow_events.append, + ) + try: + record_provider_run_started( + data_type="chip", + provider="ChipFetcher", + operation="get_chip_distribution", + ) + record_provider_run( + data_type="chip", + provider="ChipFetcher", + operation="get_chip_distribution", + success=True, + latency_ms=80, + record_count=1, + ) + finally: + reset_run_diagnostic_context(token) + + snapshot = build_task_run_flow_snapshot( + TaskInfo( + task_id="task-chip-started", + trace_id="trace-chip-started", + stock_code="600519", + stock_name="贵州茅台", + status=TaskStatus.PROCESSING, + created_at=datetime(2026, 6, 8, 10, 0, 0), + flow_events=flow_events, + ) + ) + + chip_nodes = [node for node in snapshot.nodes if node.id == "provider_chip_chipfetcher_1"] + + self.assertEqual(len(chip_nodes), 1) + self.assertEqual(chip_nodes[0].status, "success") + self.assertEqual(chip_nodes[0].record_count, 1) + self.assertEqual(chip_nodes[0].label, "筹码结构 · ChipFetcher") + self.assertIn("provider_run_started", {event.type for event in snapshot.events}) + + def test_llm_started_and_result_match_by_call_type_when_model_alias_differs(self) -> None: + flow_events: list[dict] = [] + token = activate_run_diagnostic_context( + trace_id="trace-llm-alias", + task_id="task-llm-alias", + query_id="query-llm-alias", + stock_code="600519", + trigger_source="api", + event_sink=flow_events.append, + ) + try: + record_llm_run_started( + model="deepseek-chat", + call_type="agent_analysis", + ) + record_llm_run( + success=True, + model="deepseek/deepseek-chat", + call_type="agent_analysis", + duration_ms=98000, + ) + finally: + reset_run_diagnostic_context(token) + + snapshot = build_task_run_flow_snapshot( + TaskInfo( + task_id="task-llm-alias", + trace_id="trace-llm-alias", + stock_code="600519", + stock_name="贵州茅台", + status=TaskStatus.PROCESSING, + created_at=datetime(2026, 6, 8, 10, 0, 0), + flow_events=flow_events, + ) + ) + + llm_nodes = [node for node in snapshot.nodes if node.id.startswith("llm_agent_analysis")] + + self.assertEqual([node.id for node in llm_nodes], ["llm_agent_analysis_1"]) + self.assertEqual(llm_nodes[0].status, "success") + self.assertIn("llm_run_started", {event.type for event in snapshot.events}) + self.assertIn("llm_run", {event.type for event in snapshot.events}) + + def test_completed_active_snapshot_prunes_skeleton_tail_when_live_nodes_exist(self) -> None: + flow_events: list[dict] = [] + token = activate_run_diagnostic_context( + trace_id="trace-completed-live", + task_id="task-completed-live", + query_id="query-completed-live", + stock_code="600519", + trigger_source="api", + event_sink=flow_events.append, + ) + try: + record_llm_run( + success=True, + model="deepseek/deepseek-chat", + call_type="agent_analysis", + duration_ms=98000, + ) + record_notification_run( + channel="report", + status="not_configured", + success=False, + attempts=0, + ) + finally: + reset_run_diagnostic_context(token) + + snapshot = build_task_run_flow_snapshot( + TaskInfo( + task_id="task-completed-live", + trace_id="trace-completed-live", + stock_code="600519", + stock_name="贵州茅台", + status=TaskStatus.COMPLETED, + created_at=datetime(2026, 6, 8, 10, 0, 0), + completed_at=datetime(2026, 6, 8, 10, 2, 0), + flow_events=flow_events, + ) + ) + node_ids = {node.id for node in snapshot.nodes} + + self.assertIn("llm_agent_analysis_1", node_ids) + self.assertIn("notification_report_1", node_ids) + self.assertNotIn("llm", node_ids) + self.assertNotIn("notification", node_ids) + def test_task_queue_stores_bounded_flow_events_and_broadcasts_task_progress(self) -> None: queue = AnalysisTaskQueue(max_workers=1) queue._max_flow_events_per_task = 2 @@ -713,8 +934,143 @@ class RunFlowTestCase(unittest.TestCase): self.assertEqual(snapshot.stock_code, "MARKET") self.assertEqual(snapshot.task_id, "task-market") self.assertIn("history_run", {event.type for event in snapshot.events}) + notification = next(node for node in snapshot.nodes if node.id.startswith("notification_report")) + self.assertEqual(notification.attempts, 0) self.assertTrue(snapshot.lanes) + def test_market_review_run_flow_filters_leaked_stock_provider_runs(self) -> None: + context_snapshot = { + "report_kind": "market_review", + "diagnostics": { + "trace_id": "trace-market", + "query_id": "query-flow", + "stock_code": "688521.SH", + "provider_runs": [ + { + "data_type": "daily_data", + "provider": "StockFetcher", + "success": True, + "created_at": "2026-06-13T16:00:57", + }, + { + "data_type": "news_search", + "provider": "Tavily", + "success": True, + "created_at": "2026-06-13T16:01:00", + }, + ], + "llm_runs": [], + "history_runs": [], + "notification_runs": [], + }, + } + + snapshot = build_history_run_flow_snapshot( + _history_record( + context_snapshot=context_snapshot, + code="MARKET", + name="大盘复盘", + report_type="market_review", + ) + ) + + provider_labels = {node.label for node in snapshot.nodes if node.kind == "data_source"} + self.assertEqual(snapshot.stock_code, "MARKET") + self.assertNotIn("日线K线 · StockFetcher", provider_labels) + self.assertIn("新闻舆情 · Tavily", provider_labels) + + def test_stock_run_flow_filters_nested_market_context_artifacts(self) -> None: + context_snapshot = { + "diagnostics": { + "trace_id": "trace-stock", + "query_id": "query-flow", + "stock_code": "688521.SH", + "provider_runs": [ + { + "data_type": "news_search", + "provider": "MarketNews", + "success": True, + "created_at": "2026-06-13T16:01:00", + }, + { + "data_type": "realtime_quote", + "provider": "Akshare", + "success": True, + "created_at": "2026-06-13T16:01:51", + }, + { + "data_type": "news_search", + "provider": "StockNews", + "success": True, + "created_at": "2026-06-13T16:02:25", + }, + ], + "llm_runs": [ + { + "call_type": "analysis", + "success": True, + "created_at": "2026-06-13T16:03:54", + } + ], + "history_runs": [ + { + "report_saved": True, + "metadata_saved": True, + "created_at": "2026-06-13T16:01:51", + }, + { + "report_saved": True, + "metadata_saved": True, + "created_at": "2026-06-13T16:04:12", + }, + ], + "notification_runs": [ + { + "channel": "report", + "status": "skipped", + "success": False, + "attempts": 0, + "created_at": "2026-06-13T16:01:51", + }, + { + "channel": "report", + "status": "not_configured", + "success": False, + "attempts": 0, + "created_at": "2026-06-13T16:04:12", + }, + ], + }, + "analysis_context_pack_overview": _overview( + blocks=[ + { + "key": "quote", + "label": "行情", + "status": "available", + "source": "Akshare", + "warnings": [], + "missing_reasons": [], + } + ] + ), + } + + snapshot = build_history_run_flow_snapshot(_history_record(context_snapshot=context_snapshot)) + + self.assertEqual( + [node.label for node in snapshot.nodes if node.label == "保存报告"], + ["保存报告"], + ) + self.assertEqual( + [node.label for node in snapshot.nodes if node.label.startswith("推送通知")], + ["推送通知 · report"], + ) + provider_labels = {node.label for node in snapshot.nodes if node.kind == "data_source"} + self.assertNotIn("新闻舆情 · MarketNews", provider_labels) + self.assertIn("新闻舆情 · StockNews", provider_labels) + notification = next(node for node in snapshot.nodes if node.id.startswith("notification_report")) + self.assertEqual(notification.attempts, 0) + def test_market_review_persist_records_diagnostics_without_bool_history_id(self) -> None: from src.core.market_review import _persist_market_review_history @@ -742,10 +1098,11 @@ class RunFlowTestCase(unittest.TestCase): self.assertTrue(saved) self.assertIsNotNone(fake_db.saved_context_snapshot) self.assertIn("diagnostics", fake_db.saved_context_snapshot) + self.assertIn("analysis_context_pack_overview", fake_db.saved_context_snapshot) self.assertIsNotNone(fake_db.updated_diagnostics) history_runs = fake_db.updated_diagnostics["history_runs"] self.assertTrue(history_runs) - self.assertNotEqual(history_runs[-1].get("analysis_history_id"), True) + self.assertEqual(history_runs[-1].get("analysis_history_id"), 42) def test_flow_endpoints_return_404_for_missing_records(self) -> None: with self.assertRaises(HTTPException) as history_ctx: @@ -761,6 +1118,56 @@ class RunFlowTestCase(unittest.TestCase): get_task_run_flow("missing-task") self.assertEqual(task_ctx.exception.status_code, 404) + def test_completed_task_flow_refresh_uses_persisted_history_report_type_alias(self) -> None: + task = TaskInfo( + task_id="query-flow", + trace_id="trace-flow", + stock_code="600519", + stock_name="贵州茅台", + status=TaskStatus.COMPLETED, + report_type="detailed", + ) + queue = SimpleNamespace(get_task=lambda task_id: task) + + with patch("api.v1.endpoints.analysis.get_task_queue", return_value=queue), patch( + "api.v1.endpoints.analysis._load_history_run_flow_by_query_id", + return_value=None, + ) as load_history: + snapshot = get_task_run_flow("query-flow") + + self.assertEqual(snapshot.task_id, "query-flow") + load_history.assert_called_once_with( + "query-flow", + code="600519", + report_type="full", + fail_open=True, + ) + + def test_completed_market_review_task_flow_uses_market_history_filters(self) -> None: + task = TaskInfo( + task_id="market-query-flow", + trace_id="trace-market-flow", + stock_code="cn", + stock_name="大盘复盘", + status=TaskStatus.COMPLETED, + report_type="market-review", + ) + queue = SimpleNamespace(get_task=lambda task_id: task) + + with patch("api.v1.endpoints.analysis.get_task_queue", return_value=queue), patch( + "api.v1.endpoints.analysis._load_history_run_flow_by_query_id", + return_value=None, + ) as load_history: + snapshot = get_task_run_flow("market-query-flow") + + self.assertEqual(snapshot.task_id, "market-query-flow") + load_history.assert_called_once_with( + "market-query-flow", + code="MARKET", + report_type="market_review", + fail_open=True, + ) + def test_run_flow_payload_redacts_errors_metadata_and_sensitive_paths(self) -> None: context_snapshot = { "diagnostics": _diagnostics(unsafe=True),