diff --git a/README.md b/README.md index d4b0c32..5a28791 100644 --- a/README.md +++ b/README.md @@ -248,6 +248,22 @@ The Federal Funds Rate proof should report 2 Studio field mappings, 18 annotated Open **Quality & Lineage → Lineage** to inspect both the Phase 8.1 asset graph and the Phase 8.2 field graph. Field impact is also available through `GET /api/v1/lineage/fields/impact?fieldId=...`. The field graph remains a read model: Studio-owned field mappings, dbt-owned column annotations, and dbt-owned metric expressions stay at their existing authorities. +--- + +## Phase 8.3 quick start + +Phase 8.3 projects the Phase 7 trust system onto the existing lineage nodes without creating a second lineage or quality authority. dbt checks protect their source/model and column nodes, source-controlled contract selectors protect the mart and governed mart fields, and durable active incidents follow the exact nodes they affect. + +With the API running, inspect the trust overlay: + +```powershell +Invoke-RestMethod ` + "http://localhost:8100/api/v1/lineage/trust/summary" | + ConvertTo-Json -Depth 12 +``` + +The current clean Federal Funds Rate proof should report `READY`, `TRUSTED`, and `COMPLIANT`, with 14/14 quality checks passing, 5/5 contract rules satisfied, 4 protected asset nodes, 9 protected field nodes, and zero active/blocking incidents. Open **Quality & Lineage → Lineage** to inspect the same trust evidence beside the Phase 8.1 asset and Phase 8.2 field graphs. + ## Validation contract Run the complete local validation suite before every SkyData Studio development promotion: @@ -499,7 +515,9 @@ The repository currently includes: - a completed Phase 6.1 Dockerized dbt/Postgres runtime plus the first governed staging → intermediate → mart model chain with 14 green data tests; - a completed Phase 6.2 artifact-backed dbt model catalogue surface for logical model, column, test, build, and dependency evidence; - a completed Phase 6.3 model-embedded dbt semantic definition surface with one governed semantic model, one stable entity, four dimensions, four governed metrics, and a DAY-grain MetricFlow time spine; -- a Phase 7.1 artifact-backed Data Quality surface that joins dbt test definitions to latest run outcomes and computes a consumer-facing trust posture; +- a completed Phase 7 quality stack covering dbt trust evidence, source-controlled consumer gates, durable incident lifecycle, and observation-backed reliability history; +- a completed Phase 8.1 asset/model/semantic/metric lineage graph with transitive impact analysis; +- a completed Phase 8.2 field-level lineage graph from DFF source fields through dbt derivations to governed metrics; - backend/frontend validation, repository map, and compact handoff tooling. -The active implementation target is **Phase 8.2 — Field-Level Lineage and Column Impact Foundation**. Phases 0 through 7 are complete and Phase 8.1 established the federated asset/model/semantic/metric graph. Phase 8.2 carries the same ownership model down to individual fields so schema-change impact can be traced from DFF source columns through Studio mappings, dbt derivations, and governed metrics. +The active implementation target is **Phase 8.3 — Quality and Incident Lineage Overlay Foundation**. Phases 0 through 7 are complete; Phase 8.1 established the federated asset/model/semantic/metric graph and Phase 8.2 carried that graph down to individual fields. Phase 8.3 now projects dbt quality evidence, source-controlled contract rules, and durable incident state onto the exact lineage nodes they protect without creating another authority. diff --git a/apps/api/skydata_studio/api/routes/lineage.py b/apps/api/skydata_studio/api/routes/lineage.py index d35488a..d6f3e20 100644 --- a/apps/api/skydata_studio/api/routes/lineage.py +++ b/apps/api/skydata_studio/api/routes/lineage.py @@ -9,6 +9,7 @@ FieldLineageSummary, LineageImpactSummary, LineageSummary, + LineageTrustSummary, ) from skydata_studio.services.lineage import ( field_lineage_impact, @@ -16,6 +17,7 @@ lineage_impact, lineage_summary, ) +from skydata_studio.services.lineage_trust import lineage_trust_summary router = APIRouter() @@ -72,3 +74,11 @@ def field_lineage_field_impact( return field_lineage_impact(session, field_id) except (SQLAlchemyError, OSError, ValueError, TypeError) as error: raise _unavailable(error) from error + + +@router.get("/trust/summary", response_model=LineageTrustSummary) +def lineage_trust_overlay(session: SessionDependency) -> LineageTrustSummary: + try: + return lineage_trust_summary(session) + except (SQLAlchemyError, OSError, ValueError, TypeError) as error: + raise _unavailable(error) from error diff --git a/apps/api/skydata_studio/api/routes/platform.py b/apps/api/skydata_studio/api/routes/platform.py index 69d0648..5415c2a 100644 --- a/apps/api/skydata_studio/api/routes/platform.py +++ b/apps/api/skydata_studio/api/routes/platform.py @@ -60,8 +60,8 @@ code="LINEAGE_IMPACT", name="Lineage and Impact Analysis", description=( - "Stitches metadata mappings, dbt dependencies, field lineage, semantic models, " - "and metrics into one impact graph." + "Stitches mappings, dbt and field dependencies, semantic delivery, " + "quality controls, and incidents into one impact graph." ), status="FOUNDATION", phase=8, @@ -96,7 +96,7 @@ def platform_summary() -> PlatformSummary: product="SkyData Studio", subtitle="Data Engineering Workbench", theme="Aurora Foundry", - current_phase="Phase 8.2 — Field-Level Lineage and Column Impact Foundation", + current_phase="Phase 8.3 — Quality and Incident Lineage Overlay Foundation", boundary=( "SkyData Studio starts after SkyCommand ingestion and publishes governed " "analytical products for SkyWeb Analytics, Power BI, and future consumers." diff --git a/apps/api/skydata_studio/schemas/lineage.py b/apps/api/skydata_studio/schemas/lineage.py index 9daedb4..5e2a337 100644 --- a/apps/api/skydata_studio/schemas/lineage.py +++ b/apps/api/skydata_studio/schemas/lineage.py @@ -137,3 +137,40 @@ class FieldLineageSummary(BaseModel): nodes: list[FieldLineageNode] edges: list[FieldLineageEdge] default_impact: FieldLineageImpactSummary + +LineageTrustStatus = Literal["TRUSTED", "DEGRADED", "BLOCKED", "PENDING"] + + +class LineageTrustOverlay(BaseModel): + node_id: str + node_label: str + scope: Literal["ASSET", "FIELD"] + layer: str + relation: str | None = None + quality_status: LineageTrustStatus + check_count: int = Field(ge=0) + passed_check_count: int = Field(ge=0) + warning_check_count: int = Field(ge=0) + failed_check_count: int = Field(ge=0) + contract_rule_count: int = Field(ge=0) + satisfied_contract_rule_count: int = Field(ge=0) + active_incident_count: int = Field(ge=0) + blocking_incident_count: int = Field(ge=0) + quality_dimensions: list[str] + + +class LineageTrustSummary(BaseModel): + project_name: str = "skydata_studio" + phase: str = "Phase 8.3 — Quality and Incident Lineage Overlay Foundation" + artifact_status: Literal["READY", "PARTIAL", "MISSING"] + evidence_trust_posture: Literal["TRUSTED", "DEGRADED", "BLOCKED", "PENDING"] + contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"] + check_count: int = Field(ge=0) + passed_check_count: int = Field(ge=0) + required_contract_rule_count: int = Field(ge=0) + satisfied_contract_rule_count: int = Field(ge=0) + active_incident_count: int = Field(ge=0) + blocking_incident_count: int = Field(ge=0) + protected_asset_count: int = Field(ge=0) + protected_field_count: int = Field(ge=0) + overlays: list[LineageTrustOverlay] diff --git a/apps/api/skydata_studio/services/lineage_trust.py b/apps/api/skydata_studio/services/lineage_trust.py new file mode 100644 index 0000000..4638ce6 --- /dev/null +++ b/apps/api/skydata_studio/services/lineage_trust.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +from collections import defaultdict +from typing import Literal + +from sqlalchemy.orm import Session + +from skydata_studio.schemas.lineage import ( + FieldLineageNode, + FieldLineageSummary, + LineageNode, + LineageSummary, + LineageTrustOverlay, + LineageTrustSummary, +) +from skydata_studio.schemas.quality import ( + DbtQualityCheckSummary, + DbtQualitySummary, + QualityContractRuleEvaluation, + QualityContractSummary, + QualityIncidentRead, + QualityIncidentSummary, +) +from skydata_studio.services.dbt_quality import dbt_quality_summary +from skydata_studio.services.lineage import field_lineage_summary, lineage_summary +from skydata_studio.services.quality_contracts import quality_contract_summary +from skydata_studio.services.quality_incidents import quality_incident_summary + +TrustStatus = Literal["TRUSTED", "DEGRADED", "BLOCKED", "PENDING"] + + +def _overlay_status( + checks: list[DbtQualityCheckSummary], + rules: list[QualityContractRuleEvaluation], + incidents: list[QualityIncidentRead], + *, + evidence_ready: bool, +) -> TrustStatus: + if any( + incident.severity == "BLOCKING" and incident.status != "RESOLVED" + for incident in incidents + ): + return "BLOCKED" + if any(rule.outcome in {"BLOCK", "MISSING"} for rule in rules): + return "BLOCKED" + if any(check.status in {"FAIL", "ERROR"} for check in checks): + return "BLOCKED" + if any( + incident.severity == "WARNING" and incident.status != "RESOLVED" + for incident in incidents + ): + return "DEGRADED" + if any(rule.outcome == "WARN" for rule in rules): + return "DEGRADED" + if any(check.status == "WARN" for check in checks): + return "DEGRADED" + if not evidence_ready or any(rule.outcome == "PENDING" for rule in rules): + return "PENDING" + return "TRUSTED" + + +def _quality_dimensions( + checks: list[DbtQualityCheckSummary], + rules: list[QualityContractRuleEvaluation], + incidents: list[QualityIncidentRead], +) -> list[str]: + dimensions: set[str] = set() + dimensions.update(check.quality_dimension for check in checks) + dimensions.update(rule.quality_dimension for rule in rules) + dimensions.update(incident.quality_dimension for incident in incidents) + return sorted(dimensions) + + +def _overlay( + *, + node_id: str, + node_label: str, + scope: Literal["ASSET", "FIELD"], + layer: str, + relation: str | None, + checks: list[DbtQualityCheckSummary], + rules: list[QualityContractRuleEvaluation], + incidents: list[QualityIncidentRead], + evidence_ready: bool, +) -> LineageTrustOverlay: + active_incidents = [incident for incident in incidents if incident.status != "RESOLVED"] + return LineageTrustOverlay( + node_id=node_id, + node_label=node_label, + scope=scope, + layer=layer, + relation=relation, + quality_status=_overlay_status( + checks, + rules, + incidents, + evidence_ready=evidence_ready, + ), + check_count=len(checks), + passed_check_count=sum(check.status == "PASS" for check in checks), + warning_check_count=sum(check.status == "WARN" for check in checks), + failed_check_count=sum(check.status in {"FAIL", "ERROR"} for check in checks), + contract_rule_count=len(rules), + satisfied_contract_rule_count=sum(rule.outcome == "PASS" for rule in rules), + active_incident_count=len(active_incidents), + blocking_incident_count=sum( + incident.severity == "BLOCKING" for incident in active_incidents + ), + quality_dimensions=_quality_dimensions(checks, rules, incidents), + ) + + +def compose_lineage_trust_summary( + *, + asset_lineage: LineageSummary, + field_lineage: FieldLineageSummary, + quality: DbtQualitySummary, + contract: QualityContractSummary, + incidents: QualityIncidentSummary, +) -> LineageTrustSummary: + asset_by_target: dict[str, LineageNode] = { + node.label.lower(): node + for node in asset_lineage.nodes + if node.node_type in {"DBT_SOURCE", "DBT_MODEL"} + } + field_by_target: dict[tuple[str, str], FieldLineageNode] = {} + for node in field_lineage.nodes: + if ( + node.parent_label is None + or node.node_type not in {"DBT_SOURCE_FIELD", "DBT_MODEL_FIELD"} + ): + continue + field_by_target[(node.parent_label.lower(), node.field_name.lower())] = node + + checks_by_target: dict[str, list[DbtQualityCheckSummary]] = defaultdict(list) + checks_by_field: dict[tuple[str, str], list[DbtQualityCheckSummary]] = defaultdict(list) + for check in quality.checks: + target = check.target_name.lower() + checks_by_target[target].append(check) + if check.column_name: + checks_by_field[(target, check.column_name.lower())].append(check) + + rules_by_target: dict[str, list[QualityContractRuleEvaluation]] = defaultdict(list) + rules_by_field: dict[tuple[str, str], list[QualityContractRuleEvaluation]] = defaultdict(list) + rule_by_code = {rule.code: rule for rule in contract.rules} + contract_target = contract.target_name.lower() + for rule in contract.rules: + rules_by_target[contract_target].append(rule) + if rule.column_name: + rules_by_field[(contract_target, rule.column_name.lower())].append(rule) + + incidents_by_target: dict[str, list[QualityIncidentRead]] = defaultdict(list) + incidents_by_field: dict[tuple[str, str], list[QualityIncidentRead]] = defaultdict(list) + for incident in incidents.incidents: + target = incident.target_name.lower() + incidents_by_target[target].append(incident) + incident_rule = rule_by_code.get(incident.rule_code) + if incident_rule and incident_rule.column_name: + incidents_by_field[(target, incident_rule.column_name.lower())].append(incident) + + evidence_ready = quality.artifact_status == "READY" and contract.artifact_status == "READY" + overlays: list[LineageTrustOverlay] = [] + + protected_targets = set(checks_by_target) | set(rules_by_target) | set(incidents_by_target) + for target in sorted(protected_targets): + asset_node = asset_by_target.get(target) + if asset_node is None: + continue + overlays.append( + _overlay( + node_id=asset_node.id, + node_label=asset_node.label, + scope="ASSET", + layer=asset_node.layer, + relation=asset_node.relation, + checks=checks_by_target[target], + rules=rules_by_target[target], + incidents=incidents_by_target[target], + evidence_ready=evidence_ready, + ) + ) + + protected_fields = set(checks_by_field) | set(rules_by_field) | set(incidents_by_field) + for target_field in sorted(protected_fields): + field_node = field_by_target.get(target_field) + if field_node is None: + continue + overlays.append( + _overlay( + node_id=field_node.id, + node_label=field_node.label, + scope="FIELD", + layer=field_node.layer, + relation=field_node.relation, + checks=checks_by_field[target_field], + rules=rules_by_field[target_field], + incidents=incidents_by_field[target_field], + evidence_ready=evidence_ready, + ) + ) + + if asset_lineage.artifact_status == "MISSING" and field_lineage.artifact_status == "MISSING": + artifact_status: Literal["READY", "PARTIAL", "MISSING"] = "MISSING" + elif ( + asset_lineage.artifact_status == "READY" + and field_lineage.artifact_status == "READY" + and evidence_ready + and overlays + ): + artifact_status = "READY" + else: + artifact_status = "PARTIAL" + + return LineageTrustSummary( + artifact_status=artifact_status, + evidence_trust_posture=quality.trust_posture, + contract_status=contract.contract_status, + check_count=quality.test_count, + passed_check_count=quality.passed_count, + required_contract_rule_count=contract.required_rule_count, + satisfied_contract_rule_count=contract.satisfied_rule_count, + active_incident_count=incidents.active_count, + blocking_incident_count=incidents.blocking_active_count, + protected_asset_count=sum(overlay.scope == "ASSET" for overlay in overlays), + protected_field_count=sum(overlay.scope == "FIELD" for overlay in overlays), + overlays=overlays, + ) + + +def lineage_trust_summary(session: Session) -> LineageTrustSummary: + contract = quality_contract_summary() + return compose_lineage_trust_summary( + asset_lineage=lineage_summary(session), + field_lineage=field_lineage_summary(session), + quality=dbt_quality_summary(), + contract=contract, + incidents=quality_incident_summary(session, contract=contract), + ) diff --git a/apps/web/src/App.jsx b/apps/web/src/App.jsx index 6cba61f..5ab545d 100644 --- a/apps/web/src/App.jsx +++ b/apps/web/src/App.jsx @@ -35,7 +35,7 @@ const PAGE_META = { '/quality/contracts': ['Contracts', 'Quality gates and consumer compatibility'], '/quality/incidents': ['Quality Incidents', 'Durable failures, ownership, and remediation history'], '/quality/reliability': ['Reliability', 'Quality SLO, observation history, and consumer-readiness posture'], - '/quality/lineage': ['Lineage', 'Upstream, downstream, and change impact'], + '/quality/lineage': ['Lineage', 'Asset, field, trust, and change impact'], '/delivery/marts': ['Analytical Marts', 'Curated consumer-ready facts and dimensions'], '/delivery/semantic': ['Semantic Layer', 'Governed metrics and business definitions'], '/delivery/reports': ['Reports', 'Reporting inventory and refresh evidence'], diff --git a/apps/web/src/components/LineageTrustPanel.jsx b/apps/web/src/components/LineageTrustPanel.jsx new file mode 100644 index 0000000..7dc0f7c --- /dev/null +++ b/apps/web/src/components/LineageTrustPanel.jsx @@ -0,0 +1,133 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import StatusPill from './StatusPill.jsx'; +import { getJson } from '../services/api.js'; + +const EMPTY = { + artifact_status: 'MISSING', + evidence_trust_posture: 'PENDING', + contract_status: 'PENDING', + check_count: 0, + passed_check_count: 0, + required_contract_rule_count: 0, + satisfied_contract_rule_count: 0, + active_incident_count: 0, + blocking_incident_count: 0, + protected_asset_count: 0, + protected_field_count: 0, + overlays: [], +}; + +function tone(status) { + if (status === 'TRUSTED' || status === 'COMPLIANT' || status === 'READY') return 'READY'; + if (status === 'DEGRADED' || status === 'PARTIAL') return 'WARNING'; + if (status === 'BLOCKED') return 'BLOCKED'; + return 'PLANNED'; +} + +function LineageTrustPanel() { + const [summary, setSummary] = useState(EMPTY); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(''); + + const load = useCallback(async () => { + setLoading(true); + setError(''); + try { + setSummary(await getJson('/api/v1/lineage/trust/summary')); + } catch (requestError) { + setError(requestError.message); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + const controller = new AbortController(); + getJson('/api/v1/lineage/trust/summary', { signal: controller.signal }) + .then(setSummary) + .catch((requestError) => { + if (requestError.name !== 'AbortError') setError(requestError.message); + }) + .finally(() => { + if (!controller.signal.aborted) setLoading(false); + }); + return () => controller.abort(); + }, []); + + const assets = useMemo( + () => summary.overlays.filter((overlay) => overlay.scope === 'ASSET'), + [summary.overlays], + ); + const fields = useMemo( + () => summary.overlays.filter((overlay) => overlay.scope === 'FIELD'), + [summary.overlays], + ); + + return ( +
+
+
+ PHASE 8.3 · QUALITY + INCIDENT OVERLAY +

Trust evidence follows the lineage graph

+

+ Project dbt checks, source-controlled contract rules, and durable incident state onto the + assets and fields they protect without moving ownership out of their existing systems. +

+
+
+ + +
+
+ + {error ?
!{error}
: null} + +
+
CONTRACT{summary.contract_status}{summary.satisfied_contract_rule_count}/{summary.required_contract_rule_count} rules satisfied
+
QUALITY CHECKS{summary.passed_check_count}/{summary.check_count}latest dbt evidence passing
+
PROTECTED ASSETS{summary.protected_asset_count}source + model nodes
+
PROTECTED FIELDS{summary.protected_field_count}column-level trust seams
+
ACTIVE INCIDENTS{summary.active_incident_count}{summary.blocking_incident_count} blocking
+
+ +
+
+
ASSET TRUSTdbt sources and models
{assets.length} protected
+
+ {assets.map((overlay) => ( +
+
{overlay.layer}{overlay.node_label}{overlay.quality_dimensions.join(' · ') || 'No dimensions'}
+
+ + {overlay.passed_check_count}/{overlay.check_count} checks · {overlay.satisfied_contract_rule_count}/{overlay.contract_rule_count} rules + {overlay.active_incident_count} active incident{overlay.active_incident_count === 1 ? '' : 's'} +
+
+ ))} + {!assets.length ?

No asset-level trust evidence is mapped yet.

: null} +
+
+ +
+
FIELD TRUSTquality-protected columns
{fields.length} protected
+
+ {fields.map((overlay) => ( +
+
{overlay.layer}{overlay.node_label}{overlay.quality_dimensions.join(' · ') || 'No dimensions'}
+
+ + {overlay.passed_check_count}/{overlay.check_count} checks · {overlay.satisfied_contract_rule_count}/{overlay.contract_rule_count} rules +
+
+ ))} + {!fields.length ?

No field-level trust evidence is mapped yet.

: null} +
+
+
+
+ ); +} + +export default LineageTrustPanel; diff --git a/apps/web/src/components/Sidebar.jsx b/apps/web/src/components/Sidebar.jsx index c15d392..35a9e4b 100644 --- a/apps/web/src/components/Sidebar.jsx +++ b/apps/web/src/components/Sidebar.jsx @@ -122,7 +122,7 @@ function Sidebar({ mobileOpen, onClose }) { Development - Phase 8.2 field-level lineage + Phase 8.3 trust overlay diff --git a/apps/web/src/pages/Lineage.jsx b/apps/web/src/pages/Lineage.jsx index e35c805..e278c3c 100644 --- a/apps/web/src/pages/Lineage.jsx +++ b/apps/web/src/pages/Lineage.jsx @@ -1,5 +1,6 @@ import { useCallback, useEffect, useMemo, useState } from 'react'; import FieldLineagePanel from '../components/FieldLineagePanel.jsx'; +import LineageTrustPanel from '../components/LineageTrustPanel.jsx'; import StatusPill from '../components/StatusPill.jsx'; import { getJson } from '../services/api.js'; @@ -119,11 +120,11 @@ function Lineage() {
- PHASE 8.2 · FIELD-LEVEL LINEAGE + COLUMN IMPACT + PHASE 8.3 · QUALITY + INCIDENT LINEAGE OVERLAY

Lineage & Impact

- Keep the federated asset graph from Phase 8.1, then trace field-level impact from registered - mappings through dbt derivations and into governed metrics before a schema change lands. + Preserve asset and field impact from Phases 8.1 and 8.2, then project dbt quality, contract, + and incident evidence onto the exact lineage nodes those controls protect.