Phase 7.2 — Quality Contract Gate + Consumer Compatibility

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 6 are complete. **Phase 6.3 closed green** with one governed semantic model, four metrics, one DAY-grain MetricFlow time spine, and 18/18 dbt build nodes passing. **Phase 7.1 is now in progress:** dbt test definitions and latest run outcomes become an artifact-backed Data Quality trust surface.
+**Current status:** Phases 0 through 6 are complete. **Phase 7.1 closed green** with 14/14 dbt checks passing, a TRUSTED latest-run posture, layer-aware quality evidence, and 60 local tests passing. **Phase 7.2 is now in progress:** source-controlled quality contracts turn latest dbt evidence into explicit consumer gates while the Contracts workbench keeps SkyCommand boundary compatibility visible.
 
 ---
 
@@ -161,6 +161,18 @@ Phase 6.1 materializes the proven `stg_fed_funds_rate → int_fed_funds_rate_cha
 
 Phase 7.1 keeps dbt as the test-definition and execution authority, then joins `manifest.json` test metadata with `run_results.json` outcomes through `GET /api/v1/quality/dbt/summary`. Open `/quality/checks` to inspect the latest trust posture, layer coverage, test dimensions, severity, failures, and runtime evidence. This first quality slice is observational only; persistent incidents, blocking policies, reconciliation rules, and SLOs remain later Phase 7 boundaries.
 
+## Phase 7.2 quick start
+
+Phase 7.2 adds the first source-controlled consumer quality gate under `contracts/quality/fed_funds_rate_daily.v1.json`. The policy does not copy dbt test definitions; instead, stable selectors such as target model, quality dimension, test kind, and column are evaluated against the latest Phase 7.1 dbt evidence.
+
+Core endpoint:
+
+```text
+GET /api/v1/quality/contracts/summary
+```
+
+Open `/quality/contracts` to inspect the mart gate and the existing SkyCommand consumer compatibility boundary together. The Federal Funds Rate proof requires five passing MART checks and a 100% pass rate. A missing or non-passing required rule blocks the contract; missing latest run evidence leaves it `PENDING`. Durable incident lifecycle remains a later Phase 7 slice.
+
 ---
 
 ## Validation contract

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

@@ -83,7 +83,7 @@ def platform_summary() -> PlatformSummary:
         product="SkyData Studio",
         subtitle="Data Engineering Workbench",
         theme="Aurora Foundry",
-        current_phase="Phase 7.1 — dbt Quality Evidence and Trust Posture Foundation",
+        current_phase="Phase 7.2 — Quality Contract Gate and Consumer Compatibility Workbench",
         boundary=(
             "SkyData Studio starts after SkyCommand ingestion and publishes governed "
             "analytical products for SkyWeb Analytics, Power BI, and future consumers."

apps/api/skydata_studio/api/routes/quality.py

@@ -1,20 +1,37 @@
 from fastapi import APIRouter, HTTPException, status
 
-from skydata_studio.schemas.quality import DbtQualitySummary
+from skydata_studio.schemas.quality import DbtQualitySummary, QualityContractSummary
 from skydata_studio.services.dbt_quality import dbt_quality_summary
+from skydata_studio.services.quality_contracts import quality_contract_summary
 
 router = APIRouter()
 
 
+def _artifact_unavailable(message: str) -> HTTPException:
+    return HTTPException(
+        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        detail=message,
+    )
+
+
 @router.get("/dbt/summary", response_model=DbtQualitySummary)
 def dbt_quality() -> DbtQualitySummary:
     try:
         return dbt_quality_summary()
     except (OSError, ValueError, TypeError) as error:
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail=(
-                "SkyData Studio could not read dbt quality artifacts. "
-                "Run .\\scripts\\dbt.ps1 build and refresh Data Quality."
-            ),
+        raise _artifact_unavailable(
+            "SkyData Studio could not read dbt quality artifacts. "
+            "Run .\\scripts\\dbt.ps1 build and refresh Data Quality."
+        ) from error
+
+
+@router.get("/contracts/summary", response_model=QualityContractSummary)
+def quality_contract() -> QualityContractSummary:
+    try:
+        return quality_contract_summary()
+    except (OSError, ValueError, TypeError) as error:
+        raise _artifact_unavailable(
+            "SkyData Studio could not evaluate the quality contract. "
+            "Confirm the source-controlled contract exists, run .\\scripts\\dbt.ps1 build, "
+            "and refresh Contracts."
         ) from error

apps/api/skydata_studio/schemas/quality.py

@@ -2,25 +2,29 @@
 
 from pydantic import BaseModel, Field
 
+QualityDimension = Literal[
+    "COMPLETENESS",
+    "UNIQUENESS",
+    "VALIDITY",
+    "REFERENTIAL_INTEGRITY",
+    "BUSINESS_RULE",
+    "OTHER",
+]
+QualityLayer = Literal["SOURCE", "STAGING", "INTERMEDIATE", "MART", "UNKNOWN"]
+QualityStatus = Literal["PASS", "WARN", "FAIL", "ERROR", "SKIP", "UNKNOWN"]
+
 
 class DbtQualityCheckSummary(BaseModel):
     unique_id: str
     name: str
     test_kind: Literal["GENERIC", "SINGULAR"]
-    quality_dimension: Literal[
-        "COMPLETENESS",
-        "UNIQUENESS",
-        "VALIDITY",
-        "REFERENTIAL_INTEGRITY",
-        "BUSINESS_RULE",
-        "OTHER",
-    ]
+    quality_dimension: QualityDimension
     target_name: str
     target_resource_type: Literal["MODEL", "SOURCE", "UNKNOWN"]
-    layer: Literal["SOURCE", "STAGING", "INTERMEDIATE", "MART", "UNKNOWN"]
+    layer: QualityLayer
     column_name: str | None = None
     severity: Literal["ERROR", "WARN"]
-    status: Literal["PASS", "WARN", "FAIL", "ERROR", "SKIP", "UNKNOWN"]
+    status: QualityStatus
     failures: int | None = Field(default=None, ge=0)
     execution_time_ms: float | None = Field(default=None, ge=0)
     message: str | None = None
@@ -47,3 +51,62 @@ class DbtQualitySummary(BaseModel):
     source_test_count: int = Field(ge=0)
     model_test_count: int = Field(ge=0)
     checks: list[DbtQualityCheckSummary]
+
+
+class QualityContractRuleDefinition(BaseModel):
+    code: str
+    label: str
+    quality_dimension: QualityDimension
+    test_kind: Literal["GENERIC", "SINGULAR"]
+    column_name: str | None = None
+    required_status: Literal["PASS"] = "PASS"
+
+
+class QualityContractDefinition(BaseModel):
+    code: str
+    version: str
+    name: str
+    description: str
+    target_name: str
+    layer: Literal["STAGING", "INTERMEDIATE", "MART"]
+    enforcement_mode: Literal["BLOCK", "ADVISORY"]
+    minimum_pass_rate: float = Field(ge=0.0, le=1.0)
+    rules: list[QualityContractRuleDefinition]
+
+
+class QualityContractRuleEvaluation(BaseModel):
+    code: str
+    label: str
+    quality_dimension: QualityDimension
+    test_kind: Literal["GENERIC", "SINGULAR"]
+    column_name: str | None = None
+    required_status: Literal["PASS"] = "PASS"
+    outcome: Literal["PASS", "WARN", "BLOCK", "MISSING", "PENDING"]
+    matched_check_name: str | None = None
+    matched_status: QualityStatus | None = None
+    matched_severity: Literal["ERROR", "WARN"] | None = None
+    message: str
+
+
+class QualityContractSummary(BaseModel):
+    project_name: str = "skydata_studio"
+    phase: str = "Phase 7.2 — Quality Contract Gate and Consumer Compatibility Workbench"
+    contract_code: str
+    contract_version: str
+    contract_name: str
+    description: str
+    target_name: str
+    layer: Literal["STAGING", "INTERMEDIATE", "MART"]
+    enforcement_mode: Literal["BLOCK", "ADVISORY"]
+    artifact_status: Literal["READY", "PENDING", "MISSING"]
+    evidence_trust_posture: Literal["TRUSTED", "DEGRADED", "BLOCKED", "PENDING"]
+    contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"]
+    minimum_pass_rate: float = Field(ge=0.0, le=1.0)
+    pass_rate: float = Field(ge=0.0, le=1.0)
+    required_rule_count: int = Field(ge=0)
+    satisfied_rule_count: int = Field(ge=0)
+    warning_rule_count: int = Field(ge=0)
+    blocking_rule_count: int = Field(ge=0)
+    missing_rule_count: int = Field(ge=0)
+    source_path: str
+    rules: list[QualityContractRuleEvaluation]

apps/api/skydata_studio/services/quality_contracts.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Literal, cast
+
+from skydata_studio.schemas.quality import (
+    DbtQualityCheckSummary,
+    DbtQualitySummary,
+    QualityContractDefinition,
+    QualityContractRuleDefinition,
+    QualityContractRuleEvaluation,
+    QualityContractSummary,
+)
+from skydata_studio.services.dbt_quality import dbt_quality_summary
+
+
+def _repository_root() -> Path:
+    return Path(__file__).resolve().parents[4]
+
+
+def _default_contract_path() -> Path:
+    return _repository_root() / "contracts" / "quality" / "fed_funds_rate_daily.v1.json"
+
+
+def _read_contract(path: Path) -> QualityContractDefinition:
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"Expected an object in quality contract {path}.")
+    return QualityContractDefinition.model_validate(cast(dict[str, Any], payload))
+
+
+def _matches_rule(
+    check: DbtQualityCheckSummary,
+    *,
+    contract: QualityContractDefinition,
+    rule: QualityContractRuleDefinition,
+) -> bool:
+    return (
+        check.target_name == contract.target_name
+        and check.layer == contract.layer
+        and check.quality_dimension == rule.quality_dimension
+        and check.test_kind == rule.test_kind
+        and check.column_name == rule.column_name
+    )
+
+
+def _rule_evaluation(
+    *,
+    contract: QualityContractDefinition,
+    rule: QualityContractRuleDefinition,
+    evidence: DbtQualitySummary,
+) -> QualityContractRuleEvaluation:
+    if evidence.artifact_status != "READY":
+        return QualityContractRuleEvaluation(
+            **rule.model_dump(),
+            outcome="PENDING",
+            message="Latest dbt run-result evidence is not ready yet.",
+        )
+
+    matches = [
+        check
+        for check in evidence.checks
+        if _matches_rule(check, contract=contract, rule=rule)
+    ]
+    if not matches:
+        return QualityContractRuleEvaluation(
+            **rule.model_dump(),
+            outcome="MISSING",
+            message="No latest dbt quality check satisfies this contract selector.",
+        )
+
+    check = matches[0]
+    if check.status == "PASS":
+        outcome: Literal["PASS", "WARN", "BLOCK"] = "PASS"
+        message = "Required dbt quality evidence is present and passing."
+    elif check.status == "WARN":
+        outcome = "WARN"
+        message = "Required evidence is present but the latest dbt result is warning."
+    else:
+        outcome = "BLOCK"
+        message = f"Required evidence is present but the latest dbt result is {check.status}."
+
+    if len(matches) > 1:
+        outcome = "BLOCK"
+        message = "More than one dbt check matched this contract selector; tighten the rule."
+
+    return QualityContractRuleEvaluation(
+        **rule.model_dump(),
+        outcome=outcome,
+        matched_check_name=check.name,
+        matched_status=check.status,
+        matched_severity=check.severity,
+        message=message,
+    )
+
+
+def evaluate_quality_contract(
+    contract: QualityContractDefinition,
+    evidence: DbtQualitySummary,
+    *,
+    source_path: str,
+) -> QualityContractSummary:
+    rules = [
+        _rule_evaluation(contract=contract, rule=rule, evidence=evidence)
+        for rule in contract.rules
+    ]
+    required = len(rules)
+    satisfied = sum(rule.outcome == "PASS" for rule in rules)
+    warnings = sum(rule.outcome == "WARN" for rule in rules)
+    missing = sum(rule.outcome == "MISSING" for rule in rules)
+    blocking = sum(rule.outcome in {"BLOCK", "MISSING"} for rule in rules)
+    pass_rate = satisfied / required if required else 1.0
+
+    if evidence.artifact_status != "READY":
+        contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"] = "PENDING"
+    elif blocking or pass_rate < contract.minimum_pass_rate:
+        contract_status = "BLOCKED" if contract.enforcement_mode == "BLOCK" else "DEGRADED"
+    elif warnings:
+        contract_status = "DEGRADED"
+    else:
+        contract_status = "COMPLIANT"
+
+    return QualityContractSummary(
+        contract_code=contract.code,
+        contract_version=contract.version,
+        contract_name=contract.name,
+        description=contract.description,
+        target_name=contract.target_name,
+        layer=contract.layer,
+        enforcement_mode=contract.enforcement_mode,
+        artifact_status=evidence.artifact_status,
+        evidence_trust_posture=evidence.trust_posture,
+        contract_status=contract_status,
+        minimum_pass_rate=contract.minimum_pass_rate,
+        pass_rate=pass_rate,
+        required_rule_count=required,
+        satisfied_rule_count=satisfied,
+        warning_rule_count=warnings,
+        blocking_rule_count=blocking,
+        missing_rule_count=missing,
+        source_path=source_path,
+        rules=rules,
+    )
+
+
+def quality_contract_summary(
+    *,
+    contract_path: Path | None = None,
+    target_dir: Path | None = None,
+) -> QualityContractSummary:
+    path = contract_path or _default_contract_path()
+    contract = _read_contract(path)
+    evidence = dbt_quality_summary(target_dir)
+    try:
+        source_path = path.relative_to(_repository_root()).as_posix()
+    except ValueError:
+        source_path = path.as_posix()
+    return evaluate_quality_contract(contract, evidence, source_path=source_path)

apps/web/src/App.jsx

@@ -14,6 +14,7 @@ import Transformations from './pages/Transformations.jsx';
 import DataModels from './pages/DataModels.jsx';
 import SemanticLayer from './pages/SemanticLayer.jsx';
 import DataQuality from './pages/DataQuality.jsx';
+import DataContracts from './pages/DataContracts.jsx';
 import PlaceholderPage from './pages/PlaceholderPage.jsx';
 
 const PAGE_META = {
@@ -28,7 +29,7 @@ const PAGE_META = {
   '/orchestration/runs': ['Pipeline Runs', 'Durable batch execution evidence'],
   '/orchestration/backfills': ['Schedules & Backfills', 'Time, asset, and replay controls'],
   '/quality/checks': ['Data Quality', 'Tests, incidents, policies, and trust evidence'],
-  '/quality/contracts': ['Contracts', 'Schema and consumer compatibility'],
+  '/quality/contracts': ['Contracts', 'Quality gates and consumer compatibility'],
   '/quality/lineage': ['Lineage', 'Upstream, downstream, and change impact'],
   '/delivery/marts': ['Analytical Marts', 'Curated consumer-ready facts and dimensions'],
   '/delivery/semantic': ['Semantic Layer', 'Governed metrics and business definitions'],
@@ -71,6 +72,7 @@ function App() {
           <Route path="/workspace/models" element={<DataModels />} />
           <Route path="/delivery/semantic" element={<SemanticLayer />} />
           <Route path="/quality/checks" element={<DataQuality />} />
+          <Route path="/quality/contracts" element={<DataContracts />} />
           <Route path="/orchestration/airflow" element={<Airflow />} />
           <Route path="/orchestration/runs" element={<PipelineRuns />} />
           <Route path="/orchestration/backfills" element={<SchedulesBackfills />} />
@@ -88,6 +90,7 @@ function App() {
               '/orchestration/backfills',
               '/delivery/semantic',
               '/quality/checks',
+              '/quality/contracts',
             ].includes(path))
             .map(([path, [title, subtitle]]) => (
               <Route

apps/web/src/components/Sidebar.jsx

@@ -120,7 +120,7 @@ function Sidebar({ mobileOpen, onClose }) {
         <span className="environment-dot" />
         <span>
           <strong>Development</strong>
-          <small>Phase 7.1 dbt quality evidence</small>
+          <small>Phase 7.2 quality contract gates</small>
         </span>
       </div>
     </aside>