Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 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.
**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.

---

Expand Down Expand Up @@ -171,7 +171,28 @@ Core endpoint:
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.
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`. Phase 7.2 closed with the live gate COMPLIANT at 5/5 and all five SkyCommand consumer contracts COMPATIBLE.


## Phase 7.3 quick start

Phase 7.3 adds Studio-owned runtime persistence for quality incidents without copying dbt test definitions or source-controlled contract policy. Re-run the idempotent PostgreSQL bootstrap so the two incident tables are created:

```powershell
docker compose -f .\infra\docker-compose.yml up -d studio-postgres
uv run python .\scripts\bootstrap_metadata.py
```

Then reconcile the current contract evidence into the durable register:

```text
GET /api/v1/quality/incidents/summary
POST /api/v1/quality/incidents/reconcile
POST /api/v1/quality/incidents/{incidentId}/acknowledge
POST /api/v1/quality/incidents/{incidentId}/resolve
```

Open `/quality/incidents`. A clean 5/5 contract intentionally creates **zero** incidents. `WARN`, `BLOCK`, or `MISSING` rule outcomes create one durable incident per contract rule; operators can acknowledge and manually resolve it, PASS evidence auto-resolves it, and a later recurrence reopens the same record while preserving event history. Manual resolution is not a waiver: reconciliation reopens an incident if failing evidence remains.

---

Expand Down
2 changes: 1 addition & 1 deletion apps/api/skydata_studio/api/routes/platform.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ def platform_summary() -> PlatformSummary:
product="SkyData Studio",
subtitle="Data Engineering Workbench",
theme="Aurora Foundry",
current_phase="Phase 7.2 — Quality Contract Gate and Consumer Compatibility Workbench",
current_phase="Phase 7.3 — Durable Quality Incidents and Remediation Lifecycle",
boundary=(
"SkyData Studio starts after SkyCommand ingestion and publishes governed "
"analytical products for SkyWeb Analytics, Power BI, and future consumers."
Expand Down
90 changes: 89 additions & 1 deletion apps/api/skydata_studio/api/routes/quality.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,25 @@
from fastapi import APIRouter, HTTPException, status
from sqlalchemy.exc import SQLAlchemyError

from skydata_studio.schemas.quality import DbtQualitySummary, QualityContractSummary
from skydata_studio.db.session import SessionDependency
from skydata_studio.schemas.quality import (
DbtQualitySummary,
QualityContractSummary,
QualityIncidentAction,
QualityIncidentRead,
QualityIncidentReconcileResult,
QualityIncidentSummary,
)
from skydata_studio.services.dbt_quality import dbt_quality_summary
from skydata_studio.services.quality_contracts import quality_contract_summary
from skydata_studio.services.quality_incidents import (
QualityIncidentNotFoundError,
QualityIncidentTransitionError,
acknowledge_quality_incident,
quality_incident_summary,
reconcile_quality_incidents,
resolve_quality_incident,
)

router = APIRouter()

Expand All @@ -14,6 +31,16 @@ def _artifact_unavailable(message: str) -> HTTPException:
)


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 "
"and run uv run python scripts/bootstrap_metadata.py."
),
)


@router.get("/dbt/summary", response_model=DbtQualitySummary)
def dbt_quality() -> DbtQualitySummary:
try:
Expand All @@ -35,3 +62,64 @@ def quality_contract() -> QualityContractSummary:
"Confirm the source-controlled contract exists, run .\\scripts\\dbt.ps1 build, "
"and refresh Contracts."
) from error


@router.get("/incidents/summary", response_model=QualityIncidentSummary)
def incident_summary(session: SessionDependency) -> QualityIncidentSummary:
try:
return quality_incident_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 current quality evidence for incidents. "
"Run .\\scripts\\dbt.ps1 build and retry."
) from error


@router.post("/incidents/reconcile", response_model=QualityIncidentReconcileResult)
def incident_reconcile(session: SessionDependency) -> QualityIncidentReconcileResult:
try:
return reconcile_quality_incidents(session)
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 reconcile durable incidents from current quality evidence."
) from error


@router.post("/incidents/{incident_id}/acknowledge", response_model=QualityIncidentRead)
def incident_acknowledge(
incident_id: str,
payload: QualityIncidentAction,
session: SessionDependency,
) -> QualityIncidentRead:
try:
return acknowledge_quality_incident(session, incident_id, payload)
except QualityIncidentNotFoundError as error:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(error)) from error
except QualityIncidentTransitionError as error:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(error)) from error
except SQLAlchemyError as error:
session.rollback()
raise _incident_storage_unavailable() from error


@router.post("/incidents/{incident_id}/resolve", response_model=QualityIncidentRead)
def incident_resolve(
incident_id: str,
payload: QualityIncidentAction,
session: SessionDependency,
) -> QualityIncidentRead:
try:
return resolve_quality_incident(session, incident_id, payload)
except QualityIncidentNotFoundError as error:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(error)) from error
except QualityIncidentTransitionError as error:
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(error)) from error
except SQLAlchemyError as error:
session.rollback()
raise _incident_storage_unavailable() from error
1 change: 1 addition & 0 deletions apps/api/skydata_studio/db/bootstrap.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from skydata_studio.db.session import get_engine
from skydata_studio.models import metadata as _metadata_models # noqa: F401
from skydata_studio.models import pipeline as _pipeline_models # noqa: F401
from skydata_studio.models import quality as _quality_models # noqa: F401


def create_metadata_schema(engine: Engine | None = None) -> None:
Expand Down
3 changes: 3 additions & 0 deletions apps/api/skydata_studio/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
PipelineStepRun,
PipelineVersion,
)
from skydata_studio.models.quality import QualityIncident, QualityIncidentEvent

__all__ = [
"MetadataAsset",
Expand All @@ -36,4 +37,6 @@
"PipelineStepDependency",
"PipelineStepRun",
"PipelineVersion",
"QualityIncident",
"QualityIncidentEvent",
]
75 changes: 75 additions & 0 deletions apps/api/skydata_studio/models/quality.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
from __future__ import annotations

from datetime import UTC, datetime
from uuid import uuid4

from sqlalchemy import DateTime, ForeignKey, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from skydata_studio.db.base import Base


def _uuid() -> str:
return str(uuid4())


def _utc_now() -> datetime:
return datetime.now(UTC)


class QualityIncident(Base):
__tablename__ = "quality_incident"

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
incident_key: Mapped[str] = mapped_column(String(320), unique=True, index=True)
contract_code: Mapped[str] = mapped_column(String(160), index=True)
contract_version: Mapped[str] = mapped_column(String(40))
rule_code: Mapped[str] = mapped_column(String(160), index=True)
rule_label: Mapped[str] = mapped_column(String(255))
target_name: Mapped[str] = mapped_column(String(255), index=True)
layer: Mapped[str] = mapped_column(String(40), index=True)
quality_dimension: Mapped[str] = mapped_column(String(60), index=True)
severity: Mapped[str] = mapped_column(String(40), index=True)
status: Mapped[str] = mapped_column(String(40), default="OPEN", index=True)
evidence_outcome: Mapped[str] = mapped_column(String(40))
matched_check_name: Mapped[str | None] = mapped_column(String(500))
matched_status: Mapped[str | None] = mapped_column(String(40))
message: Mapped[str] = mapped_column(Text)
occurrence_count: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
first_detected_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
last_detected_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
acknowledged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
acknowledged_by: Mapped[str | None] = mapped_column(String(160))
resolved_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
resolution_note: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=_utc_now, nullable=False
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=_utc_now, onupdate=_utc_now, nullable=False
)

events: Mapped[list[QualityIncidentEvent]] = relationship(
back_populates="incident",
cascade="all, delete-orphan",
order_by="QualityIncidentEvent.created_at",
)


class QualityIncidentEvent(Base):
__tablename__ = "quality_incident_event"

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
incident_id: Mapped[str] = mapped_column(
ForeignKey("quality_incident.id", ondelete="CASCADE"),
index=True,
)
event_type: Mapped[str] = mapped_column(String(40), index=True)
actor: Mapped[str | None] = mapped_column(String(160))
note: Mapped[str | None] = mapped_column(Text)
evidence_outcome: Mapped[str | None] = mapped_column(String(40))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=_utc_now, nullable=False
)

incident: Mapped[QualityIncident] = relationship(back_populates="events")
79 changes: 78 additions & 1 deletion apps/api/skydata_studio/schemas/quality.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from datetime import datetime
from typing import Literal

from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field

QualityDimension = Literal[
"COMPLETENESS",
Expand Down Expand Up @@ -110,3 +111,79 @@ class QualityContractSummary(BaseModel):
missing_rule_count: int = Field(ge=0)
source_path: str
rules: list[QualityContractRuleEvaluation]


QualityIncidentStatus = Literal["OPEN", "ACKNOWLEDGED", "RESOLVED"]
QualityIncidentSeverity = Literal["WARNING", "BLOCKING"]
QualityIncidentEventType = Literal["OPENED", "REOPENED", "ACKNOWLEDGED", "RESOLVED"]


class QualityIncidentEventRead(BaseModel):
model_config = ConfigDict(from_attributes=True)

id: str
event_type: QualityIncidentEventType
actor: str | None = None
note: str | None = None
evidence_outcome: Literal["PASS", "WARN", "BLOCK", "MISSING"] | None = None
created_at: datetime


class QualityIncidentRead(BaseModel):
model_config = ConfigDict(from_attributes=True)

id: str
incident_key: str
contract_code: str
contract_version: str
rule_code: str
rule_label: str
target_name: str
layer: Literal["STAGING", "INTERMEDIATE", "MART"]
quality_dimension: QualityDimension
severity: QualityIncidentSeverity
status: QualityIncidentStatus
evidence_outcome: Literal["WARN", "BLOCK", "MISSING"]
matched_check_name: str | None = None
matched_status: QualityStatus | None = None
message: str
occurrence_count: int = Field(ge=1)
first_detected_at: datetime
last_detected_at: datetime
acknowledged_at: datetime | None = None
acknowledged_by: str | None = None
resolved_at: datetime | None = None
resolution_note: str | None = None
created_at: datetime
updated_at: datetime
events: list[QualityIncidentEventRead]


class QualityIncidentAction(BaseModel):
actor: str = Field(min_length=1, max_length=160)
note: str | None = Field(default=None, max_length=2000)


class QualityIncidentSummary(BaseModel):
project_name: str = "skydata_studio"
phase: str = "Phase 7.3 — Durable Quality Incidents and Remediation Lifecycle"
contract_code: str
contract_status: Literal["COMPLIANT", "DEGRADED", "BLOCKED", "PENDING"]
artifact_status: Literal["READY", "PENDING", "MISSING"]
total_count: int = Field(ge=0)
active_count: int = Field(ge=0)
open_count: int = Field(ge=0)
acknowledged_count: int = Field(ge=0)
resolved_count: int = Field(ge=0)
blocking_active_count: int = Field(ge=0)
warning_active_count: int = Field(ge=0)
incidents: list[QualityIncidentRead]


class QualityIncidentReconcileResult(BaseModel):
project_name: str = "skydata_studio"
phase: str = "Phase 7.3 — Durable Quality Incidents and Remediation Lifecycle"
created_count: int = Field(ge=0)
reopened_count: int = Field(ge=0)
resolved_count: int = Field(ge=0)
summary: QualityIncidentSummary
Loading