diff --git a/README.md b/README.md index 8292560..e3da7b2 100644 --- a/README.md +++ b/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 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. --- @@ -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. --- diff --git a/apps/api/skydata_studio/api/routes/platform.py b/apps/api/skydata_studio/api/routes/platform.py index 53f03ed..47e50ba 100644 --- a/apps/api/skydata_studio/api/routes/platform.py +++ b/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.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." diff --git a/apps/api/skydata_studio/api/routes/quality.py b/apps/api/skydata_studio/api/routes/quality.py index a20a94c..c008c14 100644 --- a/apps/api/skydata_studio/api/routes/quality.py +++ b/apps/api/skydata_studio/api/routes/quality.py @@ -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() @@ -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: @@ -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 diff --git a/apps/api/skydata_studio/db/bootstrap.py b/apps/api/skydata_studio/db/bootstrap.py index 26c38b3..04c9da2 100644 --- a/apps/api/skydata_studio/db/bootstrap.py +++ b/apps/api/skydata_studio/db/bootstrap.py @@ -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: diff --git a/apps/api/skydata_studio/models/__init__.py b/apps/api/skydata_studio/models/__init__.py index 5cafb0a..be51af0 100644 --- a/apps/api/skydata_studio/models/__init__.py +++ b/apps/api/skydata_studio/models/__init__.py @@ -18,6 +18,7 @@ PipelineStepRun, PipelineVersion, ) +from skydata_studio.models.quality import QualityIncident, QualityIncidentEvent __all__ = [ "MetadataAsset", @@ -36,4 +37,6 @@ "PipelineStepDependency", "PipelineStepRun", "PipelineVersion", + "QualityIncident", + "QualityIncidentEvent", ] diff --git a/apps/api/skydata_studio/models/quality.py b/apps/api/skydata_studio/models/quality.py new file mode 100644 index 0000000..ecbba48 --- /dev/null +++ b/apps/api/skydata_studio/models/quality.py @@ -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") diff --git a/apps/api/skydata_studio/schemas/quality.py b/apps/api/skydata_studio/schemas/quality.py index 14ae49b..6fdbdf7 100644 --- a/apps/api/skydata_studio/schemas/quality.py +++ b/apps/api/skydata_studio/schemas/quality.py @@ -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", @@ -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 diff --git a/apps/api/skydata_studio/services/quality_incidents.py b/apps/api/skydata_studio/services/quality_incidents.py new file mode 100644 index 0000000..5ef8e48 --- /dev/null +++ b/apps/api/skydata_studio/services/quality_incidents.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Literal, cast + +from sqlalchemy import select +from sqlalchemy.orm import Session, selectinload + +from skydata_studio.models.quality import QualityIncident, QualityIncidentEvent +from skydata_studio.schemas.quality import ( + QualityContractRuleEvaluation, + QualityContractSummary, + QualityIncidentAction, + QualityIncidentRead, + QualityIncidentReconcileResult, + QualityIncidentSummary, +) +from skydata_studio.services.quality_contracts import quality_contract_summary + +IssueOutcome = Literal["WARN", "BLOCK", "MISSING"] + + +class QualityIncidentNotFoundError(ValueError): + pass + + +class QualityIncidentTransitionError(ValueError): + pass + + +def _utc_now() -> datetime: + return datetime.now(UTC) + + +def _incident_key(contract_code: str, rule_code: str) -> str: + return f"{contract_code}:{rule_code}" + + +def _issue_outcome(rule: QualityContractRuleEvaluation) -> IssueOutcome | None: + if rule.outcome in {"WARN", "BLOCK", "MISSING"}: + return cast(IssueOutcome, rule.outcome) + return None + + +def _severity(outcome: IssueOutcome) -> Literal["WARNING", "BLOCKING"]: + return "WARNING" if outcome == "WARN" else "BLOCKING" + + +def _event( + incident: QualityIncident, + event_type: Literal["OPENED", "REOPENED", "ACKNOWLEDGED", "RESOLVED"], + *, + actor: str | None, + note: str | None, + evidence_outcome: Literal["PASS", "WARN", "BLOCK", "MISSING"] | None, +) -> None: + incident.events.append( + QualityIncidentEvent( + event_type=event_type, + actor=actor, + note=note, + evidence_outcome=evidence_outcome, + ) + ) + + +def _active_incidents(session: Session, contract_code: str) -> list[QualityIncident]: + statement = ( + select(QualityIncident) + .where(QualityIncident.contract_code == contract_code) + .options(selectinload(QualityIncident.events)) + ) + return list(session.scalars(statement).unique().all()) + + +def _read(incident: QualityIncident) -> QualityIncidentRead: + return QualityIncidentRead.model_validate(incident) + + +def _summary_from_contract( + session: Session, + contract: QualityContractSummary, +) -> QualityIncidentSummary: + incidents = _active_incidents(session, contract.contract_code) + incidents.sort( + key=lambda incident: ( + incident.status == "RESOLVED", + -incident.updated_at.timestamp(), + ) + ) + active = [incident for incident in incidents if incident.status != "RESOLVED"] + return QualityIncidentSummary( + contract_code=contract.contract_code, + contract_status=contract.contract_status, + artifact_status=contract.artifact_status, + total_count=len(incidents), + active_count=len(active), + open_count=sum(incident.status == "OPEN" for incident in incidents), + acknowledged_count=sum(incident.status == "ACKNOWLEDGED" for incident in incidents), + resolved_count=sum(incident.status == "RESOLVED" for incident in incidents), + blocking_active_count=sum( + incident.status != "RESOLVED" and incident.severity == "BLOCKING" + for incident in incidents + ), + warning_active_count=sum( + incident.status != "RESOLVED" and incident.severity == "WARNING" + for incident in incidents + ), + incidents=[_read(incident) for incident in incidents], + ) + + +def quality_incident_summary( + session: Session, + *, + contract: QualityContractSummary | None = None, +) -> QualityIncidentSummary: + return _summary_from_contract(session, contract or quality_contract_summary()) + + +def reconcile_quality_incidents( + session: Session, + *, + contract: QualityContractSummary | None = None, +) -> QualityIncidentReconcileResult: + current = contract or quality_contract_summary() + incidents = _active_incidents(session, current.contract_code) + by_key = {incident.incident_key: incident for incident in incidents} + now = _utc_now() + created_count = 0 + reopened_count = 0 + resolved_count = 0 + active_rule_keys: set[str] = set() + + for rule in current.rules: + key = _incident_key(current.contract_code, rule.code) + active_rule_keys.add(key) + incident = by_key.get(key) + issue = _issue_outcome(rule) + + if issue is not None: + if incident is None: + incident = QualityIncident( + incident_key=key, + contract_code=current.contract_code, + contract_version=current.contract_version, + rule_code=rule.code, + rule_label=rule.label, + target_name=current.target_name, + layer=current.layer, + quality_dimension=rule.quality_dimension, + severity=_severity(issue), + status="OPEN", + evidence_outcome=issue, + matched_check_name=rule.matched_check_name, + matched_status=rule.matched_status, + message=rule.message, + occurrence_count=1, + first_detected_at=now, + last_detected_at=now, + ) + _event( + incident, + "OPENED", + actor="SYSTEM", + note="Latest contract evidence created a durable quality incident.", + evidence_outcome=issue, + ) + session.add(incident) + by_key[key] = incident + created_count += 1 + continue + + incident.contract_version = current.contract_version + incident.rule_label = rule.label + incident.target_name = current.target_name + incident.layer = current.layer + incident.quality_dimension = rule.quality_dimension + incident.severity = _severity(issue) + incident.evidence_outcome = issue + incident.matched_check_name = rule.matched_check_name + incident.matched_status = rule.matched_status + incident.message = rule.message + incident.last_detected_at = now + + if incident.status == "RESOLVED": + incident.status = "OPEN" + incident.occurrence_count += 1 + incident.acknowledged_at = None + incident.acknowledged_by = None + incident.resolved_at = None + incident.resolution_note = None + _event( + incident, + "REOPENED", + actor="SYSTEM", + note="Failing contract evidence returned after resolution.", + evidence_outcome=issue, + ) + reopened_count += 1 + continue + + if rule.outcome == "PASS" and incident is not None and incident.status != "RESOLVED": + incident.status = "RESOLVED" + incident.resolved_at = now + incident.resolution_note = "Latest quality contract evidence returned to PASS." + _event( + incident, + "RESOLVED", + actor="SYSTEM", + note=incident.resolution_note, + evidence_outcome="PASS", + ) + resolved_count += 1 + + for incident in incidents: + if incident.incident_key in active_rule_keys or incident.status == "RESOLVED": + continue + incident.status = "RESOLVED" + incident.resolved_at = now + incident.resolution_note = "Rule is no longer present in the active quality contract." + _event( + incident, + "RESOLVED", + actor="SYSTEM", + note=incident.resolution_note, + evidence_outcome=None, + ) + resolved_count += 1 + + session.commit() + return QualityIncidentReconcileResult( + created_count=created_count, + reopened_count=reopened_count, + resolved_count=resolved_count, + summary=_summary_from_contract(session, current), + ) + + +def _get_incident(session: Session, incident_id: str) -> QualityIncident: + statement = ( + select(QualityIncident) + .where(QualityIncident.id == incident_id) + .options(selectinload(QualityIncident.events)) + ) + incident = session.scalar(statement) + if incident is None: + raise QualityIncidentNotFoundError(f"Quality incident {incident_id} was not found.") + return incident + + +def acknowledge_quality_incident( + session: Session, + incident_id: str, + action: QualityIncidentAction, +) -> QualityIncidentRead: + incident = _get_incident(session, incident_id) + if incident.status == "RESOLVED": + raise QualityIncidentTransitionError("Resolved incidents cannot be acknowledged.") + if incident.status == "ACKNOWLEDGED": + raise QualityIncidentTransitionError("Quality incident is already acknowledged.") + + now = _utc_now() + incident.status = "ACKNOWLEDGED" + incident.acknowledged_at = now + incident.acknowledged_by = action.actor + _event( + incident, + "ACKNOWLEDGED", + actor=action.actor, + note=action.note, + evidence_outcome=cast(IssueOutcome, incident.evidence_outcome), + ) + session.commit() + session.refresh(incident) + return _read(_get_incident(session, incident.id)) + + +def resolve_quality_incident( + session: Session, + incident_id: str, + action: QualityIncidentAction, +) -> QualityIncidentRead: + incident = _get_incident(session, incident_id) + if incident.status == "RESOLVED": + raise QualityIncidentTransitionError("Quality incident is already resolved.") + + now = _utc_now() + incident.status = "RESOLVED" + incident.resolved_at = now + incident.resolution_note = action.note or "Resolved manually from SkyData Studio." + _event( + incident, + "RESOLVED", + actor=action.actor, + note=incident.resolution_note, + evidence_outcome=None, + ) + session.commit() + session.refresh(incident) + return _read(_get_incident(session, incident.id)) diff --git a/apps/web/src/App.jsx b/apps/web/src/App.jsx index 404d8cd..bc9a5b5 100644 --- a/apps/web/src/App.jsx +++ b/apps/web/src/App.jsx @@ -15,6 +15,7 @@ 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 QualityIncidents from './pages/QualityIncidents.jsx'; import PlaceholderPage from './pages/PlaceholderPage.jsx'; const PAGE_META = { @@ -28,8 +29,9 @@ const PAGE_META = { '/orchestration/airflow': ['Apache Airflow', 'DAG, asset, scheduler, and executor observability'], '/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/checks': ['Data Quality', 'Tests, coverage, and trust evidence'], '/quality/contracts': ['Contracts', 'Quality gates and consumer compatibility'], + '/quality/incidents': ['Quality Incidents', 'Durable failures, ownership, and remediation history'], '/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'], @@ -73,6 +75,7 @@ function App() { } /> } /> } /> + } /> } /> } /> } /> @@ -91,6 +94,7 @@ function App() { '/delivery/semantic', '/quality/checks', '/quality/contracts', + '/quality/incidents', ].includes(path)) .map(([path, [title, subtitle]]) => ( Development - Phase 7.2 quality contract gates + Phase 7.3 durable quality incidents diff --git a/apps/web/src/pages/DataContracts.jsx b/apps/web/src/pages/DataContracts.jsx index 4322491..a731d86 100644 --- a/apps/web/src/pages/DataContracts.jsx +++ b/apps/web/src/pages/DataContracts.jsx @@ -186,10 +186,10 @@ function DataContracts() {
-
BOUNDARY

Gate now; incidents next

▤
+
BOUNDARY

Gate established; incidents now durable

▤

- Phase 7.2 evaluates policy but does not persist incident lifecycle. Durable incidents, - acknowledgement, remediation ownership, and historical SLO evidence remain later slices. + Phase 7.2 established the policy gate. Phase 7.3 now persists incident lifecycle, + acknowledgement, remediation ownership, recurrence, and evidence-driven resolution; historical SLOs remain later work.

Code as contractArtifact evidenceNo duplicated dbt authority
diff --git a/apps/web/src/pages/DataQuality.jsx b/apps/web/src/pages/DataQuality.jsx index 125d4a8..ef34394 100644 --- a/apps/web/src/pages/DataQuality.jsx +++ b/apps/web/src/pages/DataQuality.jsx @@ -181,13 +181,13 @@ function DataQuality() {
-
BOUNDARY

Evidence first; incidents later

✓
+
BOUNDARY

Evidence first; incidents downstream

✓

- Phase 7.1 is observational. dbt remains the test-definition and execution authority; - Studio projects the latest evidence. Persistent incidents, blocking policies, SLOs, - reconciliation rules, and acknowledgements belong to later Phase 7 slices. + Phase 7.1 remains the observational evidence seam. Phase 7.2 consumes it as policy, + and Phase 7.3 now persists operational incidents downstream without changing dbt's + ownership of test definitions or execution.

-
One authorityLatest-run evidenceNo fake incidents
+
One test authorityLatest-run evidenceDownstream incident memory
diff --git a/apps/web/src/pages/QualityIncidents.jsx b/apps/web/src/pages/QualityIncidents.jsx new file mode 100644 index 0000000..6ba1268 --- /dev/null +++ b/apps/web/src/pages/QualityIncidents.jsx @@ -0,0 +1,273 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import StatusPill from '../components/StatusPill.jsx'; +import { getJson, postJson } from '../services/api.js'; + +const EMPTY = { + contract_code: 'FED_FUNDS_RATE_DAILY_QUALITY', + contract_status: 'PENDING', + artifact_status: 'PENDING', + total_count: 0, + active_count: 0, + open_count: 0, + acknowledged_count: 0, + resolved_count: 0, + blocking_active_count: 0, + warning_active_count: 0, + incidents: [], +}; + +function toneForStatus(status) { + if (['COMPLIANT', 'RESOLVED'].includes(status)) return 'READY'; + if (['DEGRADED', 'ACKNOWLEDGED', 'PENDING'].includes(status)) return 'WARNING'; + if (['BLOCKED', 'OPEN'].includes(status)) return 'BLOCKED'; + return 'UNKNOWN'; +} + +function dateLabel(value) { + if (!value) return '—'; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? value : date.toLocaleString(); +} + +function titleCase(value) { + return String(value || '—').toLowerCase().replaceAll('_', ' ').replace(/\b\w/g, (letter) => letter.toUpperCase()); +} + +function QualityIncidents() { + const [summary, setSummary] = useState(EMPTY); + const [loading, setLoading] = useState(true); + const [reconciling, setReconciling] = useState(false); + const [error, setError] = useState(''); + const [reconcileMessage, setReconcileMessage] = useState(''); + const [selectedId, setSelectedId] = useState(null); + const [actor, setActor] = useState(''); + const [note, setNote] = useState(''); + const [saving, setSaving] = useState(false); + + const selected = useMemo( + () => summary.incidents.find((incident) => incident.id === selectedId) || null, + [selectedId, summary.incidents], + ); + + const load = useCallback(async () => { + setLoading(true); + setError(''); + try { + setSummary(await getJson('/api/v1/quality/incidents/summary')); + } catch (requestError) { + setError(requestError.message); + } finally { + setLoading(false); + } + }, []); + + useEffect(() => { + const controller = new AbortController(); + getJson('/api/v1/quality/incidents/summary', { signal: controller.signal }) + .then(setSummary) + .catch((requestError) => { + if (requestError.name !== 'AbortError') setError(requestError.message); + }) + .finally(() => { + if (!controller.signal.aborted) setLoading(false); + }); + return () => controller.abort(); + }, []); + + async function reconcile() { + setReconciling(true); + setError(''); + setReconcileMessage(''); + try { + const result = await postJson('/api/v1/quality/incidents/reconcile'); + setSummary(result.summary); + setReconcileMessage( + `${result.created_count} created · ${result.reopened_count} reopened · ${result.resolved_count} auto-resolved`, + ); + } catch (requestError) { + setError(requestError.message); + } finally { + setReconciling(false); + } + } + + async function act(action) { + if (!selected || !actor.trim()) return; + setSaving(true); + setError(''); + try { + await postJson(`/api/v1/quality/incidents/${selected.id}/${action}`, { + actor: actor.trim(), + note: note.trim() || null, + }); + setSelectedId(null); + setActor(''); + setNote(''); + await load(); + } catch (requestError) { + setError(requestError.message); + } finally { + setSaving(false); + } + } + + const hasIncidents = summary.incidents.length > 0; + + return ( +
+
+
+ PHASE 7.3 · DURABLE QUALITY INCIDENTS + REMEDIATION LIFECYCLE +

Quality Incidents

+

+ Reconcile source-controlled quality gates into durable Studio-owned incident records, + then preserve acknowledgement, ownership, resolution, recurrence, and lifecycle history. +

+
+
+ + +
+
+ + {error ?
!{error}
: null} + {reconcileMessage ?
{reconcileMessage}
: null} + +
+
Contract Gate{summary.contract_status}{summary.contract_code}
+
Active{summary.active_count}{summary.open_count} open · {summary.acknowledged_count} acknowledged
+
Blocking{summary.blocking_active_count}{summary.warning_active_count} warning incidents
+
Resolved{summary.resolved_count}Durable historical records retained
+
Evidence{summary.artifact_status}Latest quality-contract evaluation
+
+ +
+
+
INCIDENT LIFECYCLE

Failure becomes owned operational evidence

+ +
+
+
01 · DETECTOPENBLOCK, MISSING, or WARN contract evidence creates one durable incident per rule.
+
02 · OWNACKNOWLEDGEDAn operator accepts responsibility with an identity and optional remediation note.
+
03 · VERIFYRESOLVEDReturning PASS evidence auto-resolves; manual resolution remains auditable.
+
04 · RECURREOPENEDA resolved rule that fails again reopens the same incident and increments occurrence history.
+
+
+ +
+
+
DURABLE REGISTER

Quality incidents and remediation ownership

+ {summary.total_count} historical incidents +
+
+ + + + + + {summary.incidents.map((incident) => ( + + + + + + + + + + + + ))} + {!hasIncidents ? ( + + + + ) : null} + +
IncidentStatusSeverityRuleEvidenceOwnerOccurrencesLast Detected
{incident.target_name}{incident.quality_dimension} · {incident.layer}{incident.severity}{incident.rule_label}{incident.rule_code}{incident.evidence_outcome}{incident.matched_status || 'No matched status'}{incident.acknowledged_by || '—'}{incident.occurrence_count}{dateLabel(incident.last_detected_at)}
+ No durable quality incidents. + The current contract is clean; reconciliation will only persist real warning or blocking evidence. +
+
+
+ + {selected ? ( +
+
+
INCIDENT DETAIL

{selected.rule_label}

+ +
+
+
Status{selected.status}{selected.severity}
+
Evidence{selected.evidence_outcome}{selected.matched_check_name || 'No matched dbt check'}
+
First Detected{dateLabel(selected.first_detected_at)}Occurrence {selected.occurrence_count}
+
Last Detected{dateLabel(selected.last_detected_at)}{selected.message}
+
+ +
+
+ +