diff --git a/js/packages/ui/src/api/index.ts b/js/packages/ui/src/api/index.ts index 8aa727144..3e38c0fd1 100644 --- a/js/packages/ui/src/api/index.ts +++ b/js/packages/ui/src/api/index.ts @@ -139,9 +139,15 @@ export type { export { submitRowCountDiff } from "./rowcount"; // Runs API (core run operations and aggregation) export type { + AggregatedRunResult, + ColumnValidationSummary, + NodeRunsAggregated, + ProfileDiffValidationSummary, RunsAggregated, SubmitOptions, SubmitRunTrackProps, + ValidationSummary, + ValueDiffValidationSummary, } from "./runs"; export { aggregateRuns, diff --git a/js/packages/ui/src/api/runs.ts b/js/packages/ui/src/api/runs.ts index 3d1434ce9..ca3fc02f7 100644 --- a/js/packages/ui/src/api/runs.ts +++ b/js/packages/ui/src/api/runs.ts @@ -6,18 +6,53 @@ import { isQueryRun, type Run, type RunType } from "./types"; // ============================================================================ /** - * Aggregated run results by model and run type + * A full result retained for legacy row-count lineage decorations. */ -export type RunsAggregated = Record< - string, - Record< - "row_count_diff" | "value_diff" | "row_count", - { - run_id: string; - result: unknown; - } - > ->; +export interface AggregatedRunResult { + run_id: string; + result: unknown; +} + +interface AvailableValidationResult { + result_available: true; +} + +export interface ValueDiffValidationSummary extends AvailableValidationResult { + latest_run_id: string; + difference_count: number; +} + +export interface ProfileDiffValidationSummary + extends AvailableValidationResult { + latest_run_id: string; + result_count: number; +} + +export interface ColumnValidationSummary extends AvailableValidationResult { + latest_run_ids_by_column: Record; + column_count: number; +} + +export interface ValidationSummary { + result_count: number; + difference_count: number; + types: { + value_diff?: ValueDiffValidationSummary; + profile_diff?: ProfileDiffValidationSummary; + top_k_diff?: ColumnValidationSummary; + histogram_diff?: ColumnValidationSummary; + }; +} + +export interface NodeRunsAggregated { + row_count_diff?: AggregatedRunResult; + value_diff?: AggregatedRunResult; + row_count?: AggregatedRunResult; + validation_summary?: ValidationSummary; +} + +/** Aggregated run results by resolved lineage node ID. */ +export type RunsAggregated = Record; /** * Properties for tracking run submissions (analytics) diff --git a/js/packages/ui/src/components/lineage/NodeTag.tsx b/js/packages/ui/src/components/lineage/NodeTag.tsx index c2dedfc47..95b3db632 100644 --- a/js/packages/ui/src/components/lineage/NodeTag.tsx +++ b/js/packages/ui/src/components/lineage/NodeTag.tsx @@ -229,7 +229,7 @@ export function RowCountDiffTag({ const { featureToggles } = useRecceInstanceContext(); const { runsAggregated } = useLineageGraphContext(); const lastRowCount: RowCountDiff | undefined = runsAggregated?.[node.id] - ?.row_count_diff.result as RowCountDiff | undefined; + ?.row_count_diff?.result as RowCountDiff | undefined; const RunTypeIcon = findByRunType("row_count_diff").icon; // Calculate during render instead of effect @@ -302,7 +302,7 @@ export function RowCountTag({ const isDark = useIsDark(); const { runsAggregated } = useLineageGraphContext(); const lastRowCount: RowCountDiff | undefined = runsAggregated?.[node.id] - ?.row_count.result as RowCountDiff | undefined; + ?.row_count?.result as RowCountDiff | undefined; const RunTypeIcon = findByRunType("row_count").icon; diff --git a/js/packages/ui/src/contexts/lineage/__tests__/LineageGraphContext.test.tsx b/js/packages/ui/src/contexts/lineage/__tests__/LineageGraphContext.test.tsx index ac2c87141..254303c94 100644 --- a/js/packages/ui/src/contexts/lineage/__tests__/LineageGraphContext.test.tsx +++ b/js/packages/ui/src/contexts/lineage/__tests__/LineageGraphContext.test.tsx @@ -93,6 +93,35 @@ function createMockRunsAggregated(): RunsAggregated { run_id: "run-3", result: 100, }, + validation_summary: { + result_count: 5, + difference_count: 2, + types: { + value_diff: { + latest_run_id: "run-4", + difference_count: 2, + result_available: true, + }, + profile_diff: { + latest_run_id: "run-5", + result_count: 1, + result_available: true, + }, + top_k_diff: { + latest_run_ids_by_column: { + status: "run-6", + region: "run-7", + }, + column_count: 2, + result_available: true, + }, + histogram_diff: { + latest_run_ids_by_column: { amount: "run-8" }, + column_count: 1, + result_available: true, + }, + }, + }, }, }; } diff --git a/js/packages/ui/src/hooks/__tests__/useRun.test.tsx b/js/packages/ui/src/hooks/__tests__/useRun.test.tsx index e648b6960..c3c7c28b2 100644 --- a/js/packages/ui/src/hooks/__tests__/useRun.test.tsx +++ b/js/packages/ui/src/hooks/__tests__/useRun.test.tsx @@ -9,7 +9,7 @@ * - New run submission (resets isRunning and completedRunIdRef) * - RunResultView component resolution * - Cancel functionality - * - Aggregated runs refetch for row_count types + * - Aggregated runs refetch for lineage aggregate run types */ import { vi } from "vitest"; @@ -596,13 +596,33 @@ describe("useRun", () => { }); }); - it("does not refetch aggregated runs for other run types", async () => { - const valueDiffRun = createMockRun({ - type: "value_diff", - result: { diff: [] }, + it.each(["value_diff", "profile_diff", "top_k_diff", "histogram_diff"])( + "refetches aggregated runs when %s completes", + async (runType) => { + const validationRun = createMockRun({ + type: runType, + result: { available: true }, + status: "Finished", + }); + mockWaitRun.mockResolvedValue(validationRun); + + renderHook(() => useRun("test-run-id"), { + wrapper: createWrapper(), + }); + + await waitFor(() => { + expect(mockRefetchRunsAggregated).toHaveBeenCalled(); + }); + }, + ); + + it("does not refetch aggregated runs for an unrelated run type", async () => { + const queryRun = createMockRun({ + type: "query", + result: { data: [] }, status: "Finished", }); - mockWaitRun.mockResolvedValue(valueDiffRun); + mockWaitRun.mockResolvedValue(queryRun); const { result } = renderHook(() => useRun("test-run-id"), { wrapper: createWrapper(), diff --git a/js/packages/ui/src/hooks/useRun.tsx b/js/packages/ui/src/hooks/useRun.tsx index e7f328452..14bafaa53 100644 --- a/js/packages/ui/src/hooks/useRun.tsx +++ b/js/packages/ui/src/hooks/useRun.tsx @@ -8,6 +8,15 @@ import { useRunsAggregated } from "../contexts"; import { useApiConfig } from "./useApiConfig"; import { useCanceledRuns } from "./useCanceledRuns"; +const AGGREGATED_RUN_TYPES = new Set([ + "row_count_diff", + "row_count", + "value_diff", + "profile_diff", + "top_k_diff", + "histogram_diff", +]); + export interface UseRunResult { run?: Run; isRunning: boolean; @@ -68,7 +77,8 @@ export const useRun = (runId?: string): UseRunResult => { useEffect(() => { if ( (error || run?.result || run?.error) && - (run?.type === "row_count_diff" || run?.type === "row_count") + run && + AGGREGATED_RUN_TYPES.has(run.type) ) { refetchRunsAggregated?.(); } diff --git a/recce/apis/run_func.py b/recce/apis/run_func.py index 4ad8520bf..33ab2fb00 100644 --- a/recce/apis/run_func.py +++ b/recce/apis/run_func.py @@ -1,12 +1,18 @@ import asyncio import logging +from collections.abc import Mapping +from datetime import timezone from typing import List, Optional, Tuple +import dateutil.parser +from pydantic import BaseModel + from recce.core import default_context from recce.exceptions import DuckDBExternalAccessBlocked, RecceException from recce.models import Run, RunDAO, RunType from recce.models.types import RunStatus from recce.tasks.core import Task +from recce.util.pydantic_model import pydantic_model_dump running_tasks = {} logger = logging.getLogger("uvicorn") @@ -328,68 +334,184 @@ def cancel_run(run_id: str) -> None: _invoke_task_cancel(task) +_VALIDATION_RUN_TYPES = ( + RunType.VALUE_DIFF, + RunType.PROFILE_DIFF, + RunType.TOP_K_DIFF, + RunType.HISTOGRAM_DIFF, +) +_COLUMN_VALIDATION_RUN_TYPES = (RunType.TOP_K_DIFF, RunType.HISTOGRAM_DIFF) + + +def _run_recency_key(run: Run): + """Return an input-order-independent ordering key for persisted runs.""" + try: + run_at = dateutil.parser.parse(run.run_at) + if run_at.tzinfo is None: + run_at = run_at.replace(tzinfo=timezone.utc) + timestamp = run_at.timestamp() + except (TypeError, ValueError, OverflowError): + timestamp = float("-inf") + return timestamp, str(run.run_id) + + +def _result_mapping(value): + """Normalize persisted mappings and live Pydantic task results.""" + if isinstance(value, Mapping): + return value + if isinstance(value, BaseModel): + try: + dumped = pydantic_model_dump(value) + except Exception: + return None + return dumped if isinstance(dumped, Mapping) else None + return None + + +def _value_diff_difference_count(result) -> Optional[int]: + """Count mismatched columns, or reject an unusable imported result.""" + result_mapping = _result_mapping(result) + if result_mapping is None: + return None + dataframe = _result_mapping(result_mapping.get("data")) + if dataframe is None: + return None + rows = dataframe.get("data") + if not isinstance(rows, (list, tuple)): + return None + + difference_count = 0 + for row in rows: + if not isinstance(row, (list, tuple)) or len(row) < 3: + return None + if row[2] is None: + continue + try: + if float(row[2]) < 1: + difference_count += 1 + except (TypeError, ValueError): + return None + return difference_count + + def materialize_run_results(runs: List[Run], nodes: List[str] = None): - """ - Materialize the run results for nodes. It walks through all runs and get the last results for primary run types. - - The result format - { - 'node_id': { - 'row_count_diff': { - 'run_id': '', - 'result': '' - }, - 'value_diff': { - 'run_id': '', - 'result': '' - }, - }, - } + """Project persisted runs into compact, node-keyed lineage evidence. + + Legacy row-count entries retain their full result shape. Validation runs + expose only IDs and scalar counts under ``validation_summary``; their raw + DataFrames and distribution arrays remain in the run store. """ context = default_context() if context: - mame_to_unique_id = context.build_name_to_unique_id_index(excluded_types={"semantic_model", "metric"}) + name_to_unique_id = context.build_name_to_unique_id_index(excluded_types={"semantic_model", "metric"}) else: - mame_to_unique_id = {} + name_to_unique_id = {} result = {} + latest_validation_runs = {} for run in runs: - if not run.result: - continue - if run.status == RunStatus.CANCELLED: - continue - - if run.type == RunType.ROW_COUNT_DIFF: + if run.type == RunType.ROW_COUNT_DIFF and run.result and run.status != RunStatus.CANCELLED: for model_name, node_run_result in run.result.items(): - key = mame_to_unique_id.get(model_name, model_name) + key = name_to_unique_id.get(model_name, model_name) if nodes: if key not in nodes: continue - if model_name not in result: - node_result = result[key] = {} - else: - node_result = result.get(key) + node_result = result.setdefault(key, {}) node_result["row_count_diff"] = { "run_id": run.run_id, "result": node_run_result, } - elif run.type == RunType.ROW_COUNT: + elif run.type == RunType.ROW_COUNT and run.result and run.status != RunStatus.CANCELLED: for model_name, node_run_result in run.result.items(): - key = mame_to_unique_id.get(model_name, model_name) + key = name_to_unique_id.get(model_name, model_name) if nodes: if key not in nodes: continue - if model_name not in result: - node_result = result[key] = {} - else: - node_result = result.get(key) + node_result = result.setdefault(key, {}) node_result["row_count"] = { "run_id": run.run_id, "result": node_run_result, } + elif run.type in _VALIDATION_RUN_TYPES and run.status == RunStatus.FINISHED and run.result is not None: + params = run.params if isinstance(run.params, dict) else {} + model_name = params.get("model") + if not isinstance(model_name, str) or not model_name: + continue + + column_name = None + if run.type in _COLUMN_VALIDATION_RUN_TYPES: + column_name = params.get("column_name") + if not isinstance(column_name, str) or not column_name: + continue + elif run.type == RunType.VALUE_DIFF and _value_diff_difference_count(run.result) is None: + continue + + key = name_to_unique_id.get(model_name, model_name) + if nodes and key not in nodes: + continue + + node_runs = latest_validation_runs.setdefault(key, {}) + if run.type in _COLUMN_VALIDATION_RUN_TYPES: + column_runs = node_runs.setdefault(run.type, {}) + previous = column_runs.get(column_name) + if previous is None or _run_recency_key(run) > _run_recency_key(previous): + column_runs[column_name] = run + else: + previous = node_runs.get(run.type) + if previous is None or _run_recency_key(run) > _run_recency_key(previous): + node_runs[run.type] = run + + for key, node_runs in latest_validation_runs.items(): + validation_types = {} + result_count = 0 + difference_count = 0 + + value_diff_run = node_runs.get(RunType.VALUE_DIFF) + if value_diff_run is not None: + value_difference_count = _value_diff_difference_count(value_diff_run.result) + if value_difference_count is not None: + difference_count = value_difference_count + validation_types[RunType.VALUE_DIFF.value] = { + "latest_run_id": value_diff_run.run_id, + "difference_count": difference_count, + "result_available": True, + } + result_count += 1 + + profile_diff_run = node_runs.get(RunType.PROFILE_DIFF) + if profile_diff_run is not None: + validation_types[RunType.PROFILE_DIFF.value] = { + "latest_run_id": profile_diff_run.run_id, + "result_count": 1, + "result_available": True, + } + result_count += 1 + + for run_type in _COLUMN_VALIDATION_RUN_TYPES: + column_runs = node_runs.get(run_type) + if not column_runs: + continue + latest_run_ids_by_column = { + column_name: column_runs[column_name].run_id for column_name in sorted(column_runs) + } + column_count = len(latest_run_ids_by_column) + validation_types[run_type.value] = { + "latest_run_ids_by_column": latest_run_ids_by_column, + "column_count": column_count, + "result_available": True, + } + result_count += column_count + + if not validation_types: + continue + result.setdefault(key, {})["validation_summary"] = { + "result_count": result_count, + "difference_count": difference_count, + "types": validation_types, + } return result diff --git a/tests/apis/test_run_func.py b/tests/apis/test_run_func.py index cf3005f63..20b52c4af 100644 --- a/tests/apis/test_run_func.py +++ b/tests/apis/test_run_func.py @@ -13,6 +13,7 @@ import contextlib import os import threading +from collections.abc import Mapping from unittest import TestCase from unittest.mock import MagicMock, patch from uuid import UUID @@ -21,7 +22,9 @@ from pydantic import BaseModel from recce.apis.run_func import materialize_run_results -from recce.state import RecceState +from recce.models.types import Run, RunStatus, RunType +from recce.state import FileStateLoader, RecceState +from recce.tasks.valuediff import ValueDiffResult current_dir = os.path.dirname(os.path.abspath(__file__)) @@ -66,6 +69,382 @@ def test_materialize_run_results_skips_cancelled_runs(): assert result == {}, "Cancelled run leaked into materialized aggregate" +def _aggregate_run( + run_type: RunType, + *, + run_id: str, + run_at: str, + result, + params=None, + status: RunStatus | None = RunStatus.FINISHED, +) -> Run: + """Build a fully specified persisted run for aggregate-contract tests.""" + return Run( + type=run_type, + params=params or {}, + result=result, + status=status, + run_at=run_at, + run_id=UUID(run_id), + ) + + +def _value_diff_result(*matched_percentages: float) -> dict: + return { + "summary": {"total": 12, "added": 1, "removed": 2}, + "data": { + "columns": [ + {"name": "column", "type": "text"}, + {"name": "matched", "type": "integer"}, + {"name": "matched_p", "type": "number"}, + ], + "data": [ + [f"column_{index}", 10, matched_percentage] + for index, matched_percentage in enumerate(matched_percentages) + ], + }, + } + + +def _assert_compact_scalars(value) -> None: + """Reject raw result containers recursively while allowing compact maps.""" + if isinstance(value, Mapping): + for nested in value.values(): + _assert_compact_scalars(nested) + return + assert isinstance(value, (bool, int, str, UUID)), f"non-compact aggregate value: {value!r}" + + +def test_materialize_validation_summaries_selects_latest_successful_results_deterministically(): + """Unsorted failed/cancelled/repeated runs cannot replace the latest usable evidence.""" + value_old_id = "00000000-0000-4000-8000-000000000001" + value_latest_id = "00000000-0000-4000-8000-000000000002" + profile_tie_lower_id = "00000000-0000-4000-8000-000000000003" + profile_tie_winner_id = "00000000-0000-4000-8000-000000000004" + top_k_status_old_id = "00000000-0000-4000-8000-000000000005" + top_k_status_latest_id = "00000000-0000-4000-8000-000000000006" + top_k_region_id = "00000000-0000-4000-8000-000000000007" + histogram_amount_id = "00000000-0000-4000-8000-000000000008" + histogram_age_id = "00000000-0000-4000-8000-000000000009" + runs = [ + _aggregate_run( + RunType.VALUE_DIFF, + run_id="00000000-0000-4000-8000-00000000000a", + run_at="2026-09-04T12:00:00Z", + status=RunStatus.CANCELLED, + params={"model": "orders"}, + result=_value_diff_result(0.0, 0.0, 0.0), + ), + _aggregate_run( + RunType.TOP_K_DIFF, + run_id=top_k_status_latest_id, + run_at="2026-09-04T10:00:00Z", + params={"model": "orders", "column_name": "status"}, + result={"base": {"values": ["paid"]}, "current": {"values": ["paid"]}}, + ), + _aggregate_run( + RunType.VALUE_DIFF, + run_id=value_old_id, + run_at="2026-09-04T08:00:00Z", + params={"model": "orders"}, + result=_value_diff_result(0.5, 0.25), + ), + _aggregate_run( + RunType.PROFILE_DIFF, + run_id=profile_tie_lower_id, + run_at="2026-09-04T11:00:00Z", + params={"model": "orders"}, + result={"base": {"data": ["raw"]}}, + ), + _aggregate_run( + RunType.HISTOGRAM_DIFF, + run_id=histogram_amount_id, + run_at="2026-09-04T07:00:00Z", + params={"model": "orders", "column_name": "amount"}, + result={"bin_edges": [0, 10], "base": {"counts": [1]}}, + ), + _aggregate_run( + RunType.VALUE_DIFF, + run_id=value_latest_id, + run_at="2026-09-04T09:00:00Z", + params={"model": "orders"}, + result=_value_diff_result(1.0, 0.75, 1.0), + ), + _aggregate_run( + RunType.TOP_K_DIFF, + run_id=top_k_status_old_id, + run_at="2026-09-04T06:00:00Z", + params={"model": "orders", "column_name": "status"}, + result={"base": {"values": ["pending"]}}, + ), + _aggregate_run( + RunType.PROFILE_DIFF, + run_id=profile_tie_winner_id, + run_at="2026-09-04T11:00:00Z", + params={"model": "orders"}, + result={}, + ), + _aggregate_run( + RunType.TOP_K_DIFF, + run_id=top_k_region_id, + run_at="2026-09-04T07:30:00Z", + params={"model": "orders", "column_name": "region"}, + result={"base": {"values": ["APAC"]}}, + ), + _aggregate_run( + RunType.HISTOGRAM_DIFF, + run_id=histogram_age_id, + run_at="2026-09-04T07:30:00Z", + params={"model": "orders", "column_name": "age"}, + result={"bin_edges": [0, 20], "current": {"counts": [3]}}, + ), + _aggregate_run( + RunType.VALUE_DIFF, + run_id="00000000-0000-4000-8000-00000000000b", + run_at="2026-09-04T13:00:00Z", + status=RunStatus.FAILED, + params={"model": "orders"}, + result=_value_diff_result(0.0, 0.0, 0.0, 0.0), + ), + _aggregate_run( + RunType.HISTOGRAM_DIFF, + run_id="00000000-0000-4000-8000-00000000000c", + run_at="2026-09-04T14:00:00Z", + status=RunStatus.RUNNING, + params={"model": "orders", "column_name": "amount"}, + result={"bin_edges": [999], "base": {"counts": [999]}}, + ), + _aggregate_run( + RunType.TOP_K_DIFF, + run_id="00000000-0000-4000-8000-00000000000d", + run_at="2026-09-04T15:00:00Z", + params={"model": "orders", "column_name": "ignored"}, + result=None, + ), + ] + + context = MagicMock() + context.build_name_to_unique_id_index.return_value = {"orders": "model.project.orders"} + with patch("recce.apis.run_func.default_context", return_value=context): + aggregated = materialize_run_results(runs) + + assert aggregated == { + "model.project.orders": { + "validation_summary": { + "result_count": 6, + "difference_count": 1, + "types": { + "value_diff": { + "latest_run_id": UUID(value_latest_id), + "difference_count": 1, + "result_available": True, + }, + "profile_diff": { + "latest_run_id": UUID(profile_tie_winner_id), + "result_count": 1, + "result_available": True, + }, + "top_k_diff": { + "latest_run_ids_by_column": { + "region": UUID(top_k_region_id), + "status": UUID(top_k_status_latest_id), + }, + "column_count": 2, + "result_available": True, + }, + "histogram_diff": { + "latest_run_ids_by_column": { + "age": UUID(histogram_age_id), + "amount": UUID(histogram_amount_id), + }, + "column_count": 2, + "result_available": True, + }, + }, + } + } + } + _assert_compact_scalars(aggregated["model.project.orders"]["validation_summary"]) + + +def test_materialize_results_uses_resolved_key_without_clobbering_and_preserves_legacy_rows(): + """Repeated status-less row-count evidence accumulates under one resolved node ID.""" + row_count_diff_id = "00000000-0000-4000-8000-000000000011" + row_count_id = "00000000-0000-4000-8000-000000000012" + validation_id = "00000000-0000-4000-8000-000000000013" + runs = [ + _aggregate_run( + RunType.ROW_COUNT_DIFF, + run_id=row_count_diff_id, + run_at="2026-09-04T08:00:00Z", + status=None, + result={"orders": {"base": 10, "curr": 12}}, + ), + _aggregate_run( + RunType.VALUE_DIFF, + run_id=validation_id, + run_at="2026-09-04T09:00:00Z", + params={"model": "orders"}, + result=_value_diff_result(1.0), + ), + _aggregate_run( + RunType.ROW_COUNT, + run_id=row_count_id, + run_at="2026-09-04T10:00:00Z", + status=None, + result={"orders": {"curr": 12}}, + ), + ] + context = MagicMock() + context.build_name_to_unique_id_index.return_value = {"orders": "model.project.orders"} + + with patch("recce.apis.run_func.default_context", return_value=context): + aggregated = materialize_run_results(runs) + + assert aggregated["model.project.orders"]["row_count_diff"] == { + "run_id": UUID(row_count_diff_id), + "result": {"base": 10, "curr": 12}, + } + assert aggregated["model.project.orders"]["row_count"] == { + "run_id": UUID(row_count_id), + "result": {"curr": 12}, + } + assert aggregated["model.project.orders"]["validation_summary"]["result_count"] == 1 + + +def test_materialize_validation_summaries_respects_resolved_node_filter(): + included_id = "00000000-0000-4000-8000-000000000021" + excluded_id = "00000000-0000-4000-8000-000000000022" + runs = [ + _aggregate_run( + RunType.PROFILE_DIFF, + run_id=included_id, + run_at="2026-09-04T08:00:00Z", + params={"model": "orders"}, + result={}, + ), + _aggregate_run( + RunType.PROFILE_DIFF, + run_id=excluded_id, + run_at="2026-09-04T08:00:00Z", + params={"model": "customers"}, + result={}, + ), + ] + context = MagicMock() + context.build_name_to_unique_id_index.return_value = { + "orders": "model.project.orders", + "customers": "model.project.customers", + } + + with patch("recce.apis.run_func.default_context", return_value=context): + aggregated = materialize_run_results(runs, nodes=["model.project.orders"]) + + assert list(aggregated) == ["model.project.orders"] + assert aggregated["model.project.orders"]["validation_summary"]["types"]["profile_diff"] == { + "latest_run_id": UUID(included_id), + "result_count": 1, + "result_available": True, + } + + +def test_materialize_validation_summary_from_file_state_loader(tmp_path): + """Persisted state runs project through the aggregate without loader changes.""" + run_id = "00000000-0000-4000-8000-000000000031" + state = RecceState( + runs=[ + _aggregate_run( + RunType.TOP_K_DIFF, + run_id=run_id, + run_at="2026-09-04T08:00:00Z", + params={"model": "orders", "column_name": "status"}, + result={"base": {"values": ["paid"]}, "current": {"values": ["paid"]}}, + ) + ] + ) + state_path = tmp_path / "persisted-recce-state.json" + state_path.write_text(state.to_json()) + loaded_state = FileStateLoader(state_file=str(state_path)).load() + + aggregated = materialize_run_results(loaded_state.runs) + + assert aggregated["orders"]["validation_summary"] == { + "result_count": 1, + "difference_count": 0, + "types": { + "top_k_diff": { + "latest_run_ids_by_column": {"status": UUID(run_id)}, + "column_count": 1, + "result_available": True, + } + }, + } + + +def test_materialize_value_diff_accepts_live_pydantic_task_result(): + """submit_run assigns ValueDiffResult after Run construction without coercing it to dict.""" + run_id = "00000000-0000-4000-8000-000000000041" + run = _aggregate_run( + RunType.VALUE_DIFF, + run_id=run_id, + run_at="2026-09-04T08:00:00Z", + params={"model": "orders"}, + result=_value_diff_result(1.0), + ) + run.result = ValueDiffResult(**_value_diff_result(1.0, 0.5, 0.25)) + + aggregated = materialize_run_results([run]) + + assert aggregated["orders"]["validation_summary"] == { + "result_count": 1, + "difference_count": 2, + "types": { + "value_diff": { + "latest_run_id": UUID(run_id), + "difference_count": 2, + "result_available": True, + } + }, + } + + +@pytest.mark.parametrize( + "malformed_result", + [ + pytest.param(["not", "a", "mapping"], id="list"), + pytest.param("not-a-result", id="string"), + pytest.param({"data": None}, id="null-dataframe"), + pytest.param({"data": {"data": "not-rows"}}, id="invalid-rows"), + ], +) +def test_materialize_value_diff_ignores_malformed_imported_results(malformed_result): + """A corrupt imported run cannot crash or claim usable lineage evidence.""" + run = _aggregate_run( + RunType.VALUE_DIFF, + run_id="00000000-0000-4000-8000-000000000042", + run_at="2026-09-04T08:00:00Z", + params={"model": "orders"}, + result=_value_diff_result(1.0), + ) + run.result = malformed_result + + assert materialize_run_results([run]) == {} + + +@pytest.mark.parametrize("run_type", [RunType.TOP_K_DIFF, RunType.HISTOGRAM_DIFF]) +def test_materialize_column_validation_without_column_name_emits_no_empty_summary(run_type): + """Malformed column-scoped imports must not create zero-result node summaries.""" + run = _aggregate_run( + run_type, + run_id="00000000-0000-4000-8000-000000000043", + run_at="2026-09-04T08:00:00Z", + params={"model": "orders"}, + result={"base": {}, "current": {}}, + ) + + assert materialize_run_results([run]) == {} + + # ============================================================================= # Integration Tests: submit_run with Mocked Task # =============================================================================