Phase 7.4 — Quality SLO + Reliability History

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 7.1 and Phase 7.2 are closed green:** latest dbt quality evidence is TRUSTED at 14/14, the Federal Funds Rate consumer gate is COMPLIANT at 5/5 and 100%, all five SkyCommand boundary contracts remain compatible, and the final Phase 7.2 local validation passed 64 tests. **Phase 7.3 is now in progress:** real non-passing contract outcomes become durable Studio-owned incidents with acknowledgement, ownership, resolution, recurrence, and lifecycle history.
+**Current status:** Phases 0 through 6 are complete. **Phase 7.1 through Phase 7.3 are closed green:** dbt quality evidence is TRUSTED at 14/14, the Federal Funds Rate consumer gate is COMPLIANT at 5/5 and 100%, durable incident reconciliation correctly creates no synthetic failure for clean evidence, and the final Phase 7.3 local validation passed 68 tests. **Phase 7.4 is now in progress:** source-controlled reliability objectives and idempotent evidence observations turn point-in-time quality gates into measurable historical SLO posture.
 
 ---
 
@@ -196,6 +196,34 @@ Open `/quality/incidents`. A clean 5/5 contract intentionally creates **zero** i
 
 ---
 
+## Phase 7.4 quick start
+
+Phase 7.4 adds an observation-backed quality SLO without pretending local captures are continuous uptime. The Federal Funds Rate quality contract now owns a 30-day, 99% minimum compliant-observation target. Re-run the idempotent metadata bootstrap so `quality_slo_observation` exists:
+
+```powershell
+docker compose -f .\infra\docker-compose.yml up -d studio-postgres
+uv run python .\scripts\bootstrap_metadata.py
+```
+
+Capture the latest quality evidence. This also reconciles durable incidents first, then writes **at most one** reliability observation for the current dbt invocation:
+
+```powershell
+Invoke-RestMethod `
+  -Method Post `
+  "http://localhost:8100/api/v1/quality/reliability/capture" |
+  ConvertTo-Json -Depth 12
+```
+
+With the current clean Federal Funds Rate evidence, the first capture should report `observation_created=true`; repeating the same capture should return `false` because the dbt invocation is already represented. Inspect the rolling posture with:
+
+```powershell
+Invoke-RestMethod `
+  "http://localhost:8100/api/v1/quality/reliability/summary" |
+  ConvertTo-Json -Depth 12
+```
+
+The first clean proof should show `MEETING`, 100% observed compliance, a 99% target, one observation, zero blocked observations, and a compliant streak of one. Open **Quality & Lineage → Reliability** for the same evidence in the workbench.
+
 ## Validation contract
 
 Run the complete local validation suite before every SkyData Studio development promotion:
@@ -306,6 +334,8 @@ Orchestration
 Quality & Lineage
   ├─ Data Quality
   ├─ Contracts
+  ├─ Quality Incidents
+  ├─ Reliability
   └─ Lineage
 
 Analytics Delivery

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.3 — Durable Quality Incidents and Remediation Lifecycle",
+        current_phase="Phase 7.4 — Quality SLO and Reliability History",
         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

@@ -9,6 +9,8 @@
     QualityIncidentRead,
     QualityIncidentReconcileResult,
     QualityIncidentSummary,
+    QualityReliabilityCaptureResult,
+    QualityReliabilitySummary,
 )
 from skydata_studio.services.dbt_quality import dbt_quality_summary
 from skydata_studio.services.quality_contracts import quality_contract_summary
@@ -20,6 +22,10 @@
     reconcile_quality_incidents,
     resolve_quality_incident,
 )
+from skydata_studio.services.quality_reliability import (
+    capture_quality_reliability_observation,
+    quality_reliability_summary,
+)
 
 router = APIRouter()
 
@@ -35,7 +41,7 @@ def _incident_storage_unavailable() -> HTTPException:
     return HTTPException(
         status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
         detail=(
-            "SkyData Studio quality incident storage is unavailable. Start studio-postgres "
+            "SkyData Studio quality operational storage is unavailable. Start studio-postgres "
             "and run uv run python scripts/bootstrap_metadata.py."
         ),
     )
@@ -80,7 +86,14 @@ def incident_summary(session: SessionDependency) -> QualityIncidentSummary:
 @router.post("/incidents/reconcile", response_model=QualityIncidentReconcileResult)
 def incident_reconcile(session: SessionDependency) -> QualityIncidentReconcileResult:
     try:
-        return reconcile_quality_incidents(session)
+        contract = quality_contract_summary()
+        result = reconcile_quality_incidents(session, contract=contract)
+        observation_created = capture_quality_reliability_observation(
+            session, contract=contract, incidents=result.summary
+        )
+        return result.model_copy(
+            update={"reliability_observation_created": observation_created}
+        )
     except SQLAlchemyError as error:
         session.rollback()
         raise _incident_storage_unavailable() from error
@@ -91,6 +104,44 @@ def incident_reconcile(session: SessionDependency) -> QualityIncidentReconcileRe
         ) from error
 
 
+@router.get("/reliability/summary", response_model=QualityReliabilitySummary)
+def reliability_summary(session: SessionDependency) -> QualityReliabilitySummary:
+    try:
+        return quality_reliability_summary(session)
+    except SQLAlchemyError as error:
+        raise _incident_storage_unavailable() from error
+    except (OSError, ValueError, TypeError) as error:
+        raise _artifact_unavailable(
+            "SkyData Studio could not evaluate quality reliability history. "
+            "Run .\\scripts\\dbt.ps1 build, reconcile quality evidence, and retry."
+        ) from error
+
+
+@router.post("/reliability/capture", response_model=QualityReliabilityCaptureResult)
+def reliability_capture(session: SessionDependency) -> QualityReliabilityCaptureResult:
+    try:
+        contract = quality_contract_summary()
+        incident_result = reconcile_quality_incidents(session, contract=contract)
+        observation_created = capture_quality_reliability_observation(
+            session, contract=contract, incidents=incident_result.summary
+        )
+        return QualityReliabilityCaptureResult(
+            observation_created=observation_created,
+            incident_created_count=incident_result.created_count,
+            incident_reopened_count=incident_result.reopened_count,
+            incident_resolved_count=incident_result.resolved_count,
+            summary=quality_reliability_summary(session, contract=contract),
+        )
+    except SQLAlchemyError as error:
+        session.rollback()
+        raise _incident_storage_unavailable() from error
+    except (OSError, ValueError, TypeError) as error:
+        session.rollback()
+        raise _artifact_unavailable(
+            "SkyData Studio could not capture the latest quality reliability observation."
+        ) from error
+
+
 @router.post("/incidents/{incident_id}/acknowledge", response_model=QualityIncidentRead)
 def incident_acknowledge(
     incident_id: str,

apps/api/skydata_studio/models/__init__.py

@@ -18,7 +18,11 @@
     PipelineStepRun,
     PipelineVersion,
 )
-from skydata_studio.models.quality import QualityIncident, QualityIncidentEvent
+from skydata_studio.models.quality import (
+    QualityIncident,
+    QualityIncidentEvent,
+    QualitySloObservation,
+)
 
 __all__ = [
     "MetadataAsset",
@@ -39,4 +43,5 @@
     "PipelineVersion",
     "QualityIncident",
     "QualityIncidentEvent",
+    "QualitySloObservation",
 ]

apps/api/skydata_studio/models/quality.py

@@ -3,7 +3,7 @@
 from datetime import UTC, datetime
 from uuid import uuid4
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, Text
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from skydata_studio.db.base import Base
@@ -73,3 +73,24 @@ class QualityIncidentEvent(Base):
     )
 
     incident: Mapped[QualityIncident] = relationship(back_populates="events")
+
+
+class QualitySloObservation(Base):
+    __tablename__ = "quality_slo_observation"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
+    observation_key: Mapped[str] = mapped_column(String(420), unique=True, index=True)
+    contract_code: Mapped[str] = mapped_column(String(160), index=True)
+    contract_version: Mapped[str] = mapped_column(String(40))
+    evidence_invocation_id: Mapped[str | None] = mapped_column(String(160), index=True)
+    evidence_generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    contract_status: Mapped[str] = mapped_column(String(40), index=True)
+    artifact_status: Mapped[str] = mapped_column(String(40), index=True)
+    evidence_trust_posture: Mapped[str] = mapped_column(String(40), index=True)
+    pass_rate: Mapped[float] = mapped_column(Float, nullable=False)
+    active_incident_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    blocking_active_incident_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    warning_active_incident_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    captured_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=_utc_now, nullable=False, index=True
+    )

apps/api/skydata_studio/schemas/quality.py

@@ -54,6 +54,11 @@ class DbtQualitySummary(BaseModel):
     checks: list[DbtQualityCheckSummary]
 
 
+class QualitySloDefinition(BaseModel):
+    window_days: int = Field(default=30, ge=1, le=365)
+    minimum_compliance_rate: float = Field(default=0.99, ge=0.0, le=1.0)
+
+
 class QualityContractRuleDefinition(BaseModel):
     code: str
     label: str
@@ -72,6 +77,7 @@ class QualityContractDefinition(BaseModel):
     layer: Literal["STAGING", "INTERMEDIATE", "MART"]
     enforcement_mode: Literal["BLOCK", "ADVISORY"]
     minimum_pass_rate: float = Field(ge=0.0, le=1.0)
+    slo: QualitySloDefinition = Field(default_factory=QualitySloDefinition)
     rules: list[QualityContractRuleDefinition]
 
 
@@ -110,6 +116,10 @@ class QualityContractSummary(BaseModel):
     blocking_rule_count: int = Field(ge=0)
     missing_rule_count: int = Field(ge=0)
     source_path: str
+    evidence_invocation_id: str | None = None
+    evidence_generated_at: str | None = None
+    slo_window_days: int = Field(default=30, ge=1, le=365)
+    slo_minimum_compliance_rate: float = Field(default=0.99, ge=0.0, le=1.0)
     rules: list[QualityContractRuleEvaluation]
 
 
@@ -186,4 +196,57 @@ class QualityIncidentReconcileResult(BaseModel):
     created_count: int = Field(ge=0)
     reopened_count: int = Field(ge=0)
     resolved_count: int = Field(ge=0)
+    reliability_observation_created: bool = False
     summary: QualityIncidentSummary
+
+
+QualityReliabilityStatus = Literal["MEETING", "AT_RISK", "BREACHED", "PENDING"]
+
+
+class QualityReliabilityObservationRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: str
+    contract_code: str
+    contract_version: str
+    evidence_invocation_id: str | None = None
+    evidence_generated_at: datetime | None = None
+    contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"]
+    artifact_status: Literal["READY", "PENDING", "MISSING"]
+    evidence_trust_posture: Literal["TRUSTED", "DEGRADED", "BLOCKED", "PENDING"]
+    pass_rate: float = Field(ge=0.0, le=1.0)
+    active_incident_count: int = Field(ge=0)
+    blocking_active_incident_count: int = Field(ge=0)
+    warning_active_incident_count: int = Field(ge=0)
+    captured_at: datetime
+
+
+class QualityReliabilitySummary(BaseModel):
+    project_name: str = "skydata_studio"
+    phase: str = "Phase 7.4 — Quality SLO and Reliability History"
+    contract_code: str
+    contract_version: str
+    current_contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"]
+    reliability_status: QualityReliabilityStatus
+    window_days: int = Field(ge=1, le=365)
+    minimum_compliance_rate: float = Field(ge=0.0, le=1.0)
+    observed_compliance_rate: float = Field(ge=0.0, le=1.0)
+    observation_count: int = Field(ge=0)
+    compliant_count: int = Field(ge=0)
+    degraded_count: int = Field(ge=0)
+    blocked_count: int = Field(ge=0)
+    pending_count: int = Field(ge=0)
+    current_compliant_streak: int = Field(ge=0)
+    window_start: datetime
+    window_end: datetime
+    observations: list[QualityReliabilityObservationRead]
+
+
+class QualityReliabilityCaptureResult(BaseModel):
+    project_name: str = "skydata_studio"
+    phase: str = "Phase 7.4 — Quality SLO and Reliability History"
+    observation_created: bool
+    incident_created_count: int = Field(ge=0)
+    incident_reopened_count: int = Field(ge=0)
+    incident_resolved_count: int = Field(ge=0)
+    summary: QualityReliabilitySummary

apps/api/skydata_studio/services/quality_contracts.py

@@ -141,6 +141,10 @@ def evaluate_quality_contract(
         blocking_rule_count=blocking,
         missing_rule_count=missing,
         source_path=source_path,
+        evidence_invocation_id=evidence.invocation_id,
+        evidence_generated_at=evidence.generated_at,
+        slo_window_days=contract.slo.window_days,
+        slo_minimum_compliance_rate=contract.slo.minimum_compliance_rate,
         rules=rules,
     )