Phase 8.2 — Field-Level Lineage + Column Impact

README.md

@@ -4,7 +4,7 @@
 
 SkyData Studio is the post-ingestion data engineering application in the Sky ecosystem. It begins where SkyCommand's ingestion responsibility ends and prepares trusted data for analytical consumption by SkyWeb Analytics, Power BI, and future client-facing applications.
 
-**Current status:** Phases 0 through 7 are complete. Phase 7 closed with TRUSTED dbt quality evidence, a COMPLIANT 5/5 consumer gate, durable incident reconciliation, and a 30-day/99% quality SLO reporting MEETING at 100% observed compliance; the final Phase 7.4 validation passed 72 tests. **Phase 8.1 is now in progress:** Studio metadata mappings, dbt dependencies, semantic models, and governed metrics are stitched into one federated lineage graph with transitive downstream impact analysis.
+**Current status:** Phases 0 through 7 and Phase 8.1 are complete. Phase 8.1 closed with a READY 11-node/10-edge federated graph, a DFF impact radius reaching all 10 downstream nodes, and final validation passing 76 tests. **Phase 8.2 is now in progress:** source-to-target field mappings, dbt column annotations, and governed metric expressions are stitched into a field-level impact graph without creating a second lineage authority.
 
 ---
 
@@ -226,21 +226,27 @@ The first clean proof should show `MEETING`, 100% observed compliance, a 99% tar
 
 ---
 
-## Phase 8.1 quick start
+## Phase 8.2 quick start
 
-Phase 8.1 composes a read-only lineage graph from existing authorities rather than creating another lineage metadata store. The first proof joins the READY `DFF → FED_FUNDS_RATE_MART` Studio mapping to the dbt source/model DAG and then continues through the semantic model into all governed metrics.
+Phase 8.2 preserves the Phase 8.1 asset graph and adds field-level lineage without asking the Studio API to guess arbitrary SQL. The two registered Studio field mappings remain the source-to-target authority, while dbt model columns declare their upstream inputs beside the model definitions in `schema.yml`; `dbt build` carries those annotations into `manifest.json`. Governed metric expressions then bind mart fields to business measures.
 
-Start Studio PostgreSQL and ensure the current dbt artifacts exist, then inspect the graph:
+Regenerate the dbt artifacts after overlaying the Phase 8.2 model annotations:
+
+```powershell
+.\scripts\dbt.ps1 build
+```
+
+Inspect the field graph:
 
 ```powershell
 Invoke-RestMethod `
-  "http://localhost:8100/api/v1/lineage/summary" |
+  "http://localhost:8100/api/v1/lineage/fields/summary" |
   ConvertTo-Json -Depth 12
 ```
 
-The current Federal Funds Rate proof should report one metadata mapping, 3 dbt business models, 1 semantic model, 4 metrics, 11 total nodes, and 10 directed edges. The default impact radius begins at `DFF` and should reach 10 downstream nodes including all 3 dbt models, the semantic model, and all 4 governed metrics.
+The Federal Funds Rate proof should report 2 Studio field mappings, 18 annotated dbt business-model columns, 4 metric bindings, 28 field/metric nodes, and 27 directed edges. Selecting `DFF.value` should reach 15 downstream nodes and exactly the average, minimum, and maximum Federal Funds Rate metrics while leaving the observation-count metric outside that radius.
 
-Open **Quality & Lineage → Lineage** and select any node to recompute its transitive downstream radius. You can query the same impact directly with `GET /api/v1/lineage/impact?nodeId=...`. Phase 8.1 remains a federated read model: field-level lineage, quality overlays, and report/consumer impact stay in later Phase 8 slices.
+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.
 
 ## Validation contract
 
@@ -496,4 +502,4 @@ The repository currently includes:
 - 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;
 - backend/frontend validation, repository map, and compact handoff tooling.
 
-The active implementation target is **Phase 8.1 — Cross-Layer Lineage Graph and Impact Radius Foundation**. Phases 0 through 7 are complete: the Studio now has governed ingestion handoffs, metadata and mappings, replay-safe pipelines, Airflow orchestration, dbt models and semantics, quality gates, durable incidents, and observation-backed reliability history. Phase 8.1 federates those existing dependency authorities into one directed graph so downstream impact can be explained without creating a second lineage store.
+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.

apps/api/skydata_studio/api/routes/lineage.py

@@ -4,8 +4,18 @@
 from sqlalchemy.exc import SQLAlchemyError
 
 from skydata_studio.db.session import SessionDependency
-from skydata_studio.schemas.lineage import LineageImpactSummary, LineageSummary
-from skydata_studio.services.lineage import lineage_impact, lineage_summary
+from skydata_studio.schemas.lineage import (
+    FieldLineageImpactSummary,
+    FieldLineageSummary,
+    LineageImpactSummary,
+    LineageSummary,
+)
+from skydata_studio.services.lineage import (
+    field_lineage_impact,
+    field_lineage_summary,
+    lineage_impact,
+    lineage_summary,
+)
 
 router = APIRouter()
 
@@ -40,3 +50,25 @@ def lineage_node_impact(
         return lineage_impact(session, node_id)
     except (SQLAlchemyError, OSError, ValueError, TypeError) as error:
         raise _unavailable(error) from error
+
+
+@router.get("/fields/summary", response_model=FieldLineageSummary)
+def field_lineage_graph(
+    session: SessionDependency,
+    focus_field_id: Annotated[str | None, Query(alias="focusFieldId")] = None,
+) -> FieldLineageSummary:
+    try:
+        return field_lineage_summary(session, focus_field_id=focus_field_id)
+    except (SQLAlchemyError, OSError, ValueError, TypeError) as error:
+        raise _unavailable(error) from error
+
+
+@router.get("/fields/impact", response_model=FieldLineageImpactSummary)
+def field_lineage_field_impact(
+    session: SessionDependency,
+    field_id: Annotated[str, Query(alias="fieldId", min_length=1)],
+) -> FieldLineageImpactSummary:
+    try:
+        return field_lineage_impact(session, field_id)
+    except (SQLAlchemyError, OSError, ValueError, TypeError) as error:
+        raise _unavailable(error) from error

apps/api/skydata_studio/api/routes/platform.py

@@ -60,7 +60,7 @@
         code="LINEAGE_IMPACT",
         name="Lineage and Impact Analysis",
         description=(
-            "Stitches metadata mappings, dbt dependencies, semantic models, "
+            "Stitches metadata mappings, dbt dependencies, field lineage, semantic models, "
             "and metrics into one impact graph."
         ),
         status="FOUNDATION",
@@ -96,7 +96,7 @@ def platform_summary() -> PlatformSummary:
         product="SkyData Studio",
         subtitle="Data Engineering Workbench",
         theme="Aurora Foundry",
-        current_phase="Phase 8.1 — Cross-Layer Lineage Graph and Impact Radius Foundation",
+        current_phase="Phase 8.2 — Field-Level Lineage and Column Impact Foundation",
         boundary=(
             "SkyData Studio starts after SkyCommand ingestion and publishes governed "
             "analytical products for SkyWeb Analytics, Power BI, and future consumers."

apps/api/skydata_studio/schemas/dbt.py

@@ -32,6 +32,7 @@ class DbtTransformationSummary(BaseModel):
 class DbtModelColumnSummary(BaseModel):
     name: str
     description: str | None = None
+    lineage_inputs: list[str] = Field(default_factory=list)
 
 
 class DbtModelDependencySummary(BaseModel):

apps/api/skydata_studio/schemas/lineage.py

@@ -74,3 +74,66 @@ class LineageSummary(BaseModel):
     nodes: list[LineageNode]
     edges: list[LineageEdge]
     default_impact: LineageImpactSummary
+
+
+FieldLineageNodeType = Literal[
+    "SOURCE_FIELD",
+    "CURATED_FIELD",
+    "DBT_SOURCE_FIELD",
+    "DBT_MODEL_FIELD",
+    "METRIC",
+]
+FieldLineageEdgeType = Literal[
+    "FIELD_MAPPING",
+    "PUBLISHES_AS",
+    "DERIVES",
+    "FEEDS_METRIC",
+]
+
+
+class FieldLineageNode(BaseModel):
+    id: str
+    label: str
+    field_name: str
+    node_type: FieldLineageNodeType
+    layer: LineageLayer
+    system: str
+    relation: str | None = None
+    parent_node_id: str | None = None
+    parent_label: str | None = None
+    description: str | None = None
+    status: LineageStatus = "UNKNOWN"
+    metadata: dict[str, str | int | float | bool | None] = Field(default_factory=dict)
+
+
+class FieldLineageEdge(BaseModel):
+    id: str
+    upstream_id: str
+    downstream_id: str
+    edge_type: FieldLineageEdgeType
+    label: str
+
+
+class FieldLineageImpactSummary(BaseModel):
+    selected_field_id: str | None = None
+    selected_field_label: str | None = None
+    downstream_node_count: int = Field(ge=0)
+    affected_field_count: int = Field(ge=0)
+    affected_metric_count: int = Field(ge=0)
+    affected_relations: list[str]
+    affected_layers: list[str]
+    nodes: list[FieldLineageNode]
+
+
+class FieldLineageSummary(BaseModel):
+    project_name: str = "skydata_studio"
+    phase: str = "Phase 8.2 — Field-Level Lineage and Column Impact Foundation"
+    artifact_status: Literal["READY", "PARTIAL", "MISSING"]
+    field_mapping_count: int = Field(ge=0)
+    dbt_annotated_column_count: int = Field(ge=0)
+    metric_binding_count: int = Field(ge=0)
+    node_count: int = Field(ge=0)
+    edge_count: int = Field(ge=0)
+    nodes: list[FieldLineageNode]
+    edges: list[FieldLineageEdge]
+    default_impact: FieldLineageImpactSummary

apps/api/skydata_studio/services/dbt_transformations.py

@@ -261,12 +261,26 @@ def dbt_model_catalogue(target_dir: Path | None = None) -> DbtModelCatalogueSumm
         if isinstance(columns_payload, dict):
             for column_name, column_payload in columns_payload.items():
                 description = None
+                lineage_inputs: list[str] = []
                 if isinstance(column_payload, dict):
                     raw_description = column_payload.get("description")
                     if isinstance(raw_description, str) and raw_description.strip():
                         description = raw_description
+                    raw_meta = column_payload.get("meta")
+                    if isinstance(raw_meta, dict):
+                        raw_lineage_inputs = raw_meta.get("lineage_inputs")
+                        if isinstance(raw_lineage_inputs, list):
+                            lineage_inputs = [
+                                str(item)
+                                for item in raw_lineage_inputs
+                                if isinstance(item, str) and item.strip()
+                            ]
                 columns.append(
-                    DbtModelColumnSummary(name=str(column_name), description=description)
+                    DbtModelColumnSummary(
+                        name=str(column_name),
+                        description=description,
+                        lineage_inputs=lineage_inputs,
+                    )
                 )
 
         test_count = 0
@@ -538,12 +552,31 @@ def dbt_semantic_layer(target_dir: Path | None = None) -> DbtSemanticLayerSummar
         metric_type = str(metric.get("type") or "simple").upper()
         if metric_type not in {"SIMPLE", "RATIO", "CUMULATIVE", "DERIVED", "CONVERSION"}:
             metric_type = "SIMPLE"
-        aggregation_value = metric.get("agg", measure.get("agg"))
-        expression_value = metric.get("expr", measure.get("expr"))
+        raw_type_params = metric.get("type_params")
+        type_params = raw_type_params if isinstance(raw_type_params, dict) else {}
+        raw_aggregation_params = type_params.get("metric_aggregation_params")
+        aggregation_params = (
+            raw_aggregation_params if isinstance(raw_aggregation_params, dict) else {}
+        )
+        aggregation_value = metric.get(
+            "agg",
+            measure.get("agg", aggregation_params.get("agg")),
+        )
+        expression_value = metric.get(
+            "expr",
+            measure.get("expr", type_params.get("expr")),
+        )
         time_dimension_value = metric.get(
             "agg_time_dimension",
-            measure.get("agg_time_dimension", default_time_dimension),
+            measure.get(
+                "agg_time_dimension",
+                aggregation_params.get("agg_time_dimension", default_time_dimension),
+            ),
         )
+        if semantic_name is None:
+            raw_semantic_name = aggregation_params.get("semantic_model")
+            if isinstance(raw_semantic_name, str) and raw_semantic_name:
+                semantic_name = raw_semantic_name
         aggregation = (
             str(aggregation_value) if isinstance(aggregation_value, str) else None
         )