Phase 8.4 — Pipeline + Airflow Execution Lineage Foundation

README.md

@@ -518,6 +518,22 @@ The repository currently includes:
 - 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;
+- a completed Phase 8.3 trust overlay that projects dbt checks, quality-contract rules, and durable incidents onto exact lineage nodes;
+- a Phase 8.4 runtime-lineage workbench that links the latest Airflow DAG/task execution to its replay-safe Studio pipeline run, structured steps, source asset, and curated target;
 - backend/frontend validation, repository map, and compact handoff tooling.
 
-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.
+The active implementation target is **Phase 8.4 — Pipeline and Airflow Execution Lineage Foundation**. Phases 0 through 7 are complete; Phase 8.1 established asset-level impact, Phase 8.2 added field-level impact, and Phase 8.3 projected trust evidence onto those nodes. Phase 8.4 now joins the latest Airflow DAG/task proof to the replay-safe Studio pipeline run and its structured materialization steps without reading Airflow metadata tables directly.
+
+
+## Phase 8.4 quick start
+
+Phase 8.4 federates runtime evidence into lineage without moving execution ownership. Airflow remains authoritative for DAG-run and task-instance state; Studio remains authoritative for pipeline-run and step evidence. The runtime graph links the existing `DFF` and `FED_FUNDS_RATE_MART` structural nodes to the latest `AIRFLOW:` Studio run.
+
+```powershell
+# Ensure the Airflow proof runtime is available, then start the Studio API/web as usual.
+Invoke-RestMethod `
+  "http://localhost:8100/api/v1/lineage/runtime/summary" |
+  ConvertTo-Json -Depth 12
+```
+
+A fully connected proof reports `runtime_status=READY`, four successful Airflow tasks, four successful Studio steps, and replay-safe materialization evidence for `mart.fed_funds_rate`.

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

@@ -1,22 +1,25 @@
 from typing import Annotated
 
-from fastapi import APIRouter, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.exc import SQLAlchemyError
 
+from skydata_studio.core.config import Settings, get_settings
 from skydata_studio.db.session import SessionDependency
 from skydata_studio.schemas.lineage import (
     FieldLineageImpactSummary,
     FieldLineageSummary,
     LineageImpactSummary,
     LineageSummary,
     LineageTrustSummary,
+    RuntimeLineageSummary,
 )
 from skydata_studio.services.lineage import (
     field_lineage_impact,
     field_lineage_summary,
     lineage_impact,
     lineage_summary,
 )
+from skydata_studio.services.lineage_runtime import runtime_lineage_summary
 from skydata_studio.services.lineage_trust import lineage_trust_summary
 
 router = APIRouter()
@@ -82,3 +85,14 @@ def lineage_trust_overlay(session: SessionDependency) -> LineageTrustSummary:
         return lineage_trust_summary(session)
     except (SQLAlchemyError, OSError, ValueError, TypeError) as error:
         raise _unavailable(error) from error
+
+
+@router.get("/runtime/summary", response_model=RuntimeLineageSummary)
+def lineage_runtime_execution(
+    session: SessionDependency,
+    settings: Annotated[Settings, Depends(get_settings)],
+) -> RuntimeLineageSummary:
+    try:
+        return runtime_lineage_summary(session, settings)
+    except (SQLAlchemyError, OSError, ValueError, TypeError) as error:
+        raise _unavailable(error) from error

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

@@ -61,7 +61,7 @@
         name="Lineage and Impact Analysis",
         description=(
             "Stitches mappings, dbt and field dependencies, semantic delivery, "
-            "quality controls, and incidents into one impact graph."
+            "quality controls, incidents, and runtime execution 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.3 — Quality and Incident Lineage Overlay Foundation",
+        current_phase="Phase 8.4 — Pipeline and Airflow Execution Lineage 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/lineage.py

@@ -174,3 +174,70 @@ class LineageTrustSummary(BaseModel):
     protected_asset_count: int = Field(ge=0)
     protected_field_count: int = Field(ge=0)
     overlays: list[LineageTrustOverlay]
+
+
+RuntimeLineageNodeType = Literal[
+    "STRUCTURAL_ASSET",
+    "PIPELINE_DEFINITION",
+    "AIRFLOW_DAG",
+    "AIRFLOW_DAG_RUN",
+    "AIRFLOW_TASK",
+    "STUDIO_PIPELINE_RUN",
+    "STUDIO_STEP_RUN",
+]
+RuntimeLineageEdgeType = Literal[
+    "READS_FROM",
+    "ORCHESTRATED_BY",
+    "EXECUTION",
+    "TASK_FLOW",
+    "CALLS_STUDIO",
+    "STEP_FLOW",
+    "MATERIALIZES",
+]
+
+
+class RuntimeLineageNode(BaseModel):
+    id: str
+    label: str
+    node_type: RuntimeLineageNodeType
+    system: str
+    status: str
+    relation: str | None = None
+    metadata: dict[str, object] = Field(default_factory=dict)
+
+
+class RuntimeLineageEdge(BaseModel):
+    id: str
+    upstream_id: str
+    downstream_id: str
+    edge_type: RuntimeLineageEdgeType
+    label: str
+
+
+class RuntimeLineageSummary(BaseModel):
+    project_name: str = "skydata_studio"
+    phase: str = "Phase 8.4 — Pipeline and Airflow Execution Lineage Foundation"
+    runtime_status: Literal["READY", "PARTIAL", "MISSING"]
+    airflow_connection_status: Literal["CONNECTED", "UNAVAILABLE", "UNKNOWN"]
+    pipeline_code: str
+    dag_id: str
+    dag_run_id: str | None = None
+    studio_run_id: str | None = None
+    studio_run_key: str | None = None
+    airflow_dag_run_status: str | None = None
+    studio_run_status: str | None = None
+    airflow_task_count: int = Field(default=0, ge=0)
+    successful_airflow_task_count: int = Field(default=0, ge=0)
+    studio_step_count: int = Field(default=0, ge=0)
+    succeeded_studio_step_count: int = Field(default=0, ge=0)
+    replay_count: int = Field(default=0, ge=0)
+    materialization_executed: bool = False
+    data_mutation_applied: bool = False
+    target_relation: str | None = None
+    target_row_count: int | None = Field(default=None, ge=0)
+    airflow_error: str | None = None
+    node_count: int = Field(default=0, ge=0)
+    edge_count: int = Field(default=0, ge=0)
+    nodes: list[RuntimeLineageNode] = Field(default_factory=list)
+    edges: list[RuntimeLineageEdge] = Field(default_factory=list)
+