diff --git a/semapact/reconciliation/__init__.py b/semapact/reconciliation/__init__.py new file mode 100644 index 0000000..8550bf6 --- /dev/null +++ b/semapact/reconciliation/__init__.py @@ -0,0 +1,19 @@ +"""Platform-neutral governed-desired-vs-observed reconciliation.""" + +from semapact.reconciliation.engine import reconcile_governed_contract +from semapact.reconciliation.models import ( + ReconciliationDifference, + ReconciliationDifferenceType, + ReconciliationResult, + ReconciliationSubject, + serialize_reconciliation_result, +) + +__all__ = [ + "ReconciliationDifference", + "ReconciliationDifferenceType", + "ReconciliationResult", + "ReconciliationSubject", + "reconcile_governed_contract", + "serialize_reconciliation_result", +] diff --git a/semapact/reconciliation/engine.py b/semapact/reconciliation/engine.py new file mode 100644 index 0000000..9dc5b2d --- /dev/null +++ b/semapact/reconciliation/engine.py @@ -0,0 +1,296 @@ +"""Deterministic governed-desired-state to observed-state reconciliation.""" + +from __future__ import annotations + +from open_data_contract_standard.model import OpenDataContractStandard, SchemaProperty + +from semapact.exceptions import ValidationError +from semapact.lifecycle.identity import ( + PropertyIdentity, + build_property_index, + build_schema_index, + normalize_identity_name, +) +from semapact.observation.fingerprint import fingerprint_observed_state +from semapact.observation.models import ObservedAsset, ObservedPlatformState, ObservedProperty +from semapact.reconciliation.models import ( + ReconciliationDifference, + ReconciliationDifferenceType, + ReconciliationResult, + ReconciliationSubject, +) + +_SUBJECT_ORDER = { + ReconciliationSubject.ASSET: 0, + ReconciliationSubject.PROPERTY: 1, + ReconciliationSubject.PHYSICAL_TYPE: 2, + ReconciliationSubject.NULLABILITY: 3, +} + + +def reconcile_governed_contract( + contract: OpenDataContractStandard, + observation: ObservedPlatformState, +) -> ReconciliationResult: + """Compare governed ODCS desired state with platform-neutral observed state. + + The caller is responsible for selecting the authoritative governed contract + revision. Reconciliation reports raw differences only; it does not determine + approval/authorization, classify drift cause or operational status, or mutate + either input. + """ + governed_assets = build_schema_index(contract) + observed_assets = _build_observed_asset_index(observation) + differences: list[ReconciliationDifference] = [] + + governed_keys = set(governed_assets) + observed_keys = set(observed_assets) + + for asset_key in sorted(governed_keys - observed_keys): + differences.append( + _difference( + difference_type=ReconciliationDifferenceType.MISSING, + subject=ReconciliationSubject.ASSET, + asset_identity=asset_key, + ) + ) + + for asset_key in sorted(observed_keys - governed_keys): + differences.append( + _difference( + difference_type=ReconciliationDifferenceType.UNEXPECTED, + subject=ReconciliationSubject.ASSET, + asset_identity=asset_key, + ) + ) + + for asset_key in sorted(governed_keys & observed_keys): + governed_schema = governed_assets[asset_key] + observed_asset = observed_assets[asset_key] + differences.extend( + _reconcile_properties( + asset_key=asset_key, + governed_properties=list(governed_schema.properties or []), + observed_asset=observed_asset, + ) + ) + + ordered = tuple(sorted(differences, key=_difference_sort_key)) + return ReconciliationResult( + contract_id=_required_contract_text(getattr(contract, "id", None), field="id"), + contract_version=_required_contract_text( + getattr(contract, "version", None), field="version" + ), + observation_source_identifier=observation.source_identifier, + observation_fingerprint=( + observation.fingerprint or fingerprint_observed_state(observation) + ), + differences=ordered, + ) + + +def _reconcile_properties( + *, + asset_key: str, + governed_properties: list[SchemaProperty], + observed_asset: ObservedAsset, +) -> list[ReconciliationDifference]: + governed = build_property_index(asset_key, governed_properties) + observed = _build_observed_property_index(asset_key, observed_asset) + differences: list[ReconciliationDifference] = [] + + governed_keys = set(governed) + observed_keys = set(observed) + + for prop_key in sorted(governed_keys - observed_keys): + differences.append( + _difference( + difference_type=ReconciliationDifferenceType.MISSING, + subject=ReconciliationSubject.PROPERTY, + asset_identity=asset_key, + property_identity=prop_key[1], + ) + ) + + for prop_key in sorted(observed_keys - governed_keys): + differences.append( + _difference( + difference_type=ReconciliationDifferenceType.UNEXPECTED, + subject=ReconciliationSubject.PROPERTY, + asset_identity=asset_key, + property_identity=prop_key[1], + ) + ) + + for prop_key in sorted(governed_keys & observed_keys): + differences.extend( + _reconcile_matching_property( + asset_key=asset_key, + property_key=prop_key, + governed=governed[prop_key], + observed=observed[prop_key], + ) + ) + + return differences + + +def _reconcile_matching_property( + *, + asset_key: str, + property_key: PropertyIdentity, + governed: SchemaProperty, + observed: ObservedProperty, +) -> list[ReconciliationDifference]: + differences: list[ReconciliationDifference] = [] + property_identity = property_key[1] + + expected_physical = _optional_text(getattr(governed, "physicalType", None)) + observed_physical = _optional_text(observed.physical_type) + if ( + expected_physical is not None + and observed_physical is not None + and _normalize_comparable_text(expected_physical) + != _normalize_comparable_text(observed_physical) + ): + differences.append( + _difference( + difference_type=ReconciliationDifferenceType.MISMATCH, + subject=ReconciliationSubject.PHYSICAL_TYPE, + asset_identity=asset_key, + property_identity=property_identity, + expected=expected_physical, + observed=observed_physical, + ) + ) + + required = getattr(governed, "required", None) + nullable = observed.nullable + if isinstance(required, bool) and isinstance(nullable, bool): + expected_nullable = not required + if expected_nullable != nullable: + differences.append( + _difference( + difference_type=ReconciliationDifferenceType.MISMATCH, + subject=ReconciliationSubject.NULLABILITY, + asset_identity=asset_key, + property_identity=property_identity, + expected=expected_nullable, + observed=nullable, + ) + ) + + return differences + + +def _build_observed_asset_index( + observation: ObservedPlatformState, +) -> dict[str, ObservedAsset]: + index: dict[str, ObservedAsset] = {} + for asset in observation.assets: + key = normalize_identity_name(asset.identity.asset, "Observed asset") + if key in index: + raise ValidationError( + f"Duplicate canonical observed asset identity found: '{key}'" + ) + index[key] = asset + return index + + +def _build_observed_property_index( + asset_key: str, + asset: ObservedAsset, +) -> dict[PropertyIdentity, ObservedProperty]: + index: dict[PropertyIdentity, ObservedProperty] = {} + for prop in asset.properties: + if prop.identity.asset != asset.identity: + raise ValidationError( + "Observed property asset identity must match its containing asset" + ) + prop_name = normalize_identity_name( + prop.identity.property, "Observed property" + ) + key: PropertyIdentity = (asset_key, prop_name) + if key in index: + raise ValidationError( + f"Duplicate canonical observed property identity found: '{prop_name}'" + f" in asset '{asset_key}'" + ) + index[key] = prop + return index + + +def _difference( + *, + difference_type: ReconciliationDifferenceType, + subject: ReconciliationSubject, + asset_identity: str, + property_identity: str | None = None, + expected: str | bool | None = None, + observed: str | bool | None = None, +) -> ReconciliationDifference: + return ReconciliationDifference( + difference_type=difference_type, + subject=subject, + path=_difference_path( + subject=subject, + asset_identity=asset_identity, + property_identity=property_identity, + ), + asset_identity=asset_identity, + property_identity=property_identity, + expected=expected, + observed=observed, + ) + + +def _difference_path( + *, + subject: ReconciliationSubject, + asset_identity: str, + property_identity: str | None, +) -> str: + asset_path = f"schema[{asset_identity}]" + if subject is ReconciliationSubject.ASSET: + return asset_path + + if property_identity is None: + raise ValueError(f"property_identity is required for {subject.value}") + + property_path = f"{asset_path}.properties[{property_identity}]" + if subject is ReconciliationSubject.PROPERTY: + return property_path + if subject is ReconciliationSubject.PHYSICAL_TYPE: + return f"{property_path}.physicalType" + return f"{property_path}.nullability" + + +def _difference_sort_key( + difference: ReconciliationDifference, +) -> tuple[str, str, int, str]: + return ( + difference.asset_identity, + difference.property_identity or "", + _SUBJECT_ORDER[difference.subject], + difference.difference_type.value, + ) + + +def _required_contract_text(value: object, *, field: str) -> str: + text = _optional_text(value) + if text is None: + raise ValidationError( + f"Governed desired-state contract {field} is required for reconciliation" + ) + return text + + +def _optional_text(value: object | None) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +def _normalize_comparable_text(value: str) -> str: + return value.strip().casefold() diff --git a/semapact/reconciliation/models.py b/semapact/reconciliation/models.py new file mode 100644 index 0000000..8538e27 --- /dev/null +++ b/semapact/reconciliation/models.py @@ -0,0 +1,73 @@ +"""Platform-neutral reconciliation result models. + +Reconciliation reports raw desired-vs-observed differences. It does not infer +drift cause, deployment state, or governance status; those classifications are +separate downstream concerns. +""" + +from __future__ import annotations + +import json +from enum import Enum + +from pydantic import BaseModel, ConfigDict + + +class ReconciliationModel(BaseModel): + """Shared immutable base for reconciliation models.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + +class ReconciliationDifferenceType(str, Enum): + """Generic raw comparison operation, not a public reason-code taxonomy.""" + + MISSING = "missing" + UNEXPECTED = "unexpected" + MISMATCH = "mismatch" + + +class ReconciliationSubject(str, Enum): + """Comparable subject represented by a raw reconciliation difference.""" + + ASSET = "asset" + PROPERTY = "property" + PHYSICAL_TYPE = "physical_type" + NULLABILITY = "nullability" + + +class ReconciliationDifference(ReconciliationModel): + """One deterministic raw difference between desired and observed state.""" + + difference_type: ReconciliationDifferenceType + subject: ReconciliationSubject + path: str + asset_identity: str + property_identity: str | None = None + expected: str | bool | None = None + observed: str | bool | None = None + + +class ReconciliationResult(ReconciliationModel): + """Raw desired-vs-observed comparison result.""" + + contract_id: str + contract_version: str + observation_source_identifier: str + observation_fingerprint: str + differences: tuple[ReconciliationDifference, ...] = () + + @property + def has_differences(self) -> bool: + """Return whether any raw differences were found.""" + return bool(self.differences) + + +def serialize_reconciliation_result(result: ReconciliationResult) -> str: + """Serialize a reconciliation result deterministically for machine use.""" + return json.dumps( + result.model_dump(mode="json"), + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ) diff --git a/tests/test_reconciliation.py b/tests/test_reconciliation.py new file mode 100644 index 0000000..7b57069 --- /dev/null +++ b/tests/test_reconciliation.py @@ -0,0 +1,300 @@ +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime, timezone + +import pytest +from open_data_contract_standard.model import ( + OpenDataContractStandard, + SchemaObject, + SchemaProperty, +) + +from semapact.exceptions import ValidationError +from semapact.observation import ( + ObservedAsset, + ObservedAssetIdentity, + ObservedPlatformState, + ObservedProperty, + ObservedPropertyIdentity, + with_observed_state_fingerprint, +) +from semapact.reconciliation import ( + ReconciliationDifferenceType, + ReconciliationSubject, + reconcile_governed_contract, + serialize_reconciliation_result, +) + +CAPTURED_AT = datetime(2026, 9, 1, 10, 0, tzinfo=timezone.utc) + + +def _contract(*schemas: SchemaObject) -> OpenDataContractStandard: + return OpenDataContractStandard.model_construct( + id="orders-contract", + version="1.2.3", + schema_=list(schemas), + ) + + +def _property( + *, + asset: ObservedAssetIdentity, + name: str, + physical_type: str | None, + nullable: bool | None, +) -> ObservedProperty: + return ObservedProperty( + identity=ObservedPropertyIdentity(asset=asset, property=name), + physical_type=physical_type, + nullable=nullable, + ) + + +def _asset( + name: str, + *properties: tuple[str, str | None, bool | None], + namespace: tuple[str, ...] = ("main", "silver"), +) -> ObservedAsset: + identity = ObservedAssetIdentity( + platform="databricks", + namespace=namespace, + asset=name, + ) + return ObservedAsset( + identity=identity, + properties=tuple( + _property( + asset=identity, + name=prop_name, + physical_type=physical_type, + nullable=nullable, + ) + for prop_name, physical_type, nullable in properties + ), + ) + + +def _observation(*assets: ObservedAsset) -> ObservedPlatformState: + state = ObservedPlatformState( + platform="databricks", + source_identifier="https://adb.example", + assets=tuple(assets), + captured_at=CAPTURED_AT, + fingerprint=None, + ) + return with_observed_state_fingerprint(state) + + +def test_exact_comparable_state_has_no_differences() -> None: + contract = _contract( + SchemaObject( + name=" Orders ", + properties=[ + SchemaProperty( + name="Order_ID", + type="integer", + physicalType="BIGINT", + required=True, + ), + SchemaProperty( + name="amount", + type="number", + physicalType="DECIMAL(18,2)", + required=False, + ), + ], + ) + ) + observation = _observation( + _asset( + "orders", + ("order_id", "bigint", False), + ("amount", "decimal(18,2)", True), + namespace=("other_catalog", "other_schema"), + ) + ) + + result = reconcile_governed_contract(contract, observation) + + assert result.differences == () + assert result.has_differences is False + assert result.contract_id == "orders-contract" + assert result.contract_version == "1.2.3" + assert result.observation_fingerprint == observation.fingerprint + + +def test_missing_and_unexpected_assets_and_properties_are_reported() -> None: + contract = _contract( + SchemaObject( + name="orders", + properties=[ + SchemaProperty(name="id", type="integer", physicalType="BIGINT"), + SchemaProperty(name="amount", type="number", physicalType="DOUBLE"), + ], + ), + SchemaObject(name="customers", properties=[]), + ) + observation = _observation( + _asset( + "orders", + ("id", "bigint", None), + ("note", "string", True), + ), + _asset("payments"), + ) + + result = reconcile_governed_contract(contract, observation) + + assert [ + (item.difference_type, item.subject, item.path) + for item in result.differences + ] == [ + ( + ReconciliationDifferenceType.MISSING, + ReconciliationSubject.ASSET, + "schema[customers]", + ), + ( + ReconciliationDifferenceType.MISSING, + ReconciliationSubject.PROPERTY, + "schema[orders].properties[amount]", + ), + ( + ReconciliationDifferenceType.UNEXPECTED, + ReconciliationSubject.PROPERTY, + "schema[orders].properties[note]", + ), + ( + ReconciliationDifferenceType.UNEXPECTED, + ReconciliationSubject.ASSET, + "schema[payments]", + ), + ] + + +def test_physical_type_and_nullability_mismatches_are_reported() -> None: + contract = _contract( + SchemaObject( + name="orders", + properties=[ + SchemaProperty( + name="customer_id", + type="integer", + physicalType="BIGINT", + required=True, + ) + ], + ) + ) + observation = _observation( + _asset("orders", ("customer_id", "STRING", True)) + ) + + result = reconcile_governed_contract(contract, observation) + + assert len(result.differences) == 2 + physical, nullable = result.differences + assert physical.subject is ReconciliationSubject.PHYSICAL_TYPE + assert physical.expected == "BIGINT" + assert physical.observed == "STRING" + assert nullable.subject is ReconciliationSubject.NULLABILITY + assert nullable.expected is False + assert nullable.observed is True + + +def test_unknown_comparable_values_are_not_guessed() -> None: + contract = _contract( + SchemaObject( + name="orders", + properties=[SchemaProperty(name="id", type="integer")], + ) + ) + observation = _observation(_asset("orders", ("id", None, None))) + + result = reconcile_governed_contract(contract, observation) + + assert result.differences == () + + +def test_duplicate_canonical_observed_asset_identity_fails_closed() -> None: + contract = _contract(SchemaObject(name="orders", properties=[])) + observation = _observation( + _asset("orders", namespace=("main", "one")), + _asset(" ORDERS ", namespace=("main", "two")), + ) + + with pytest.raises( + ValidationError, + match="Duplicate canonical observed asset identity found: 'orders'", + ): + reconcile_governed_contract(contract, observation) + + +def test_duplicate_canonical_observed_property_identity_fails_closed() -> None: + contract = _contract( + SchemaObject( + name="orders", + properties=[SchemaProperty(name="id", type="integer")], + ) + ) + observation = _observation( + _asset( + "orders", + ("id", "bigint", False), + (" ID ", "bigint", False), + ) + ) + + with pytest.raises( + ValidationError, + match="Duplicate canonical observed property identity found: 'id'", + ): + reconcile_governed_contract(contract, observation) + + +def test_difference_order_and_serialization_are_deterministic() -> None: + orders = SchemaObject( + name="orders", + properties=[ + SchemaProperty(name="z_col", type="string", physicalType="STRING"), + SchemaProperty(name="a_col", type="integer", physicalType="BIGINT"), + ], + ) + users = SchemaObject(name="users", properties=[]) + contract_left = _contract(users, orders) + contract_right = _contract(orders, users) + + orders_observed = _asset( + "orders", + ("extra", "string", True), + ("a_col", "string", None), + ) + payments_observed = _asset("payments") + observation_left = _observation(payments_observed, orders_observed) + observation_right = _observation(orders_observed, payments_observed) + + left = reconcile_governed_contract(contract_left, observation_left) + right = reconcile_governed_contract(contract_right, observation_right) + + assert left.differences == right.differences + assert serialize_reconciliation_result(left) == serialize_reconciliation_result(right) + + +def test_reconciliation_does_not_mutate_inputs() -> None: + contract = _contract( + SchemaObject( + name="orders", + properties=[ + SchemaProperty(name="id", type="integer", physicalType="BIGINT") + ], + ) + ) + observation = _observation(_asset("orders", ("id", "string", False))) + contract_before = deepcopy(contract.model_dump()) + observation_before = observation.model_dump() + + reconcile_governed_contract(contract, observation) + + assert contract.model_dump() == contract_before + assert observation.model_dump() == observation_before