DataOps-platform/app/core/governance/domain_replication.py

"""Generic, evidence-bound acceptance for copying a governance domain.

The evaluator consumes a self-contained implementation package.  It validates
controlled snapshot/delta ingestion and canonical receipts from existing
platform modules; it does not execute domain-specific business logic or store
source rows in the report.
"""

from __future__ import annotations

import hashlib
import json
import re
from copy import deepcopy
from typing import Any

REQUIRED_STAGES = (
    "template_initialization",
    "incremental_ingestion",
    "catalog",
    "semantics",
    "responsibility",
    "quality",
    "remediation",
    "observability",
    "data_product",
    "agent",
)

STAGE_SUBSYSTEMS = {
    "template_initialization": "domain_templates",
    "incremental_ingestion": "active_metadata",
    "catalog": "active_metadata",
    "semantics": "semantic_governance",
    "responsibility": "unified_responsibilities",
    "quality": "quality_operations",
    "remediation": "unified_work_center",
    "observability": "data_observability",
    "data_product": "product_governance",
    "agent": "agent_governance",
}

_CODE = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
_SECRET = re.compile(
    r"(?:password|passwd|secret|token|credential|authorization|api[_-]?key)",
    re.IGNORECASE,
)
_DEVICE_PATH_MARKERS = (
    "app/core/data_research/device_",
    "app/api/data_development/device_",
    "frontend/src/views/dataResearch/device",
)
_REQUIRED_ENTERPRISE_BINDINGS = {
    "data_steward",
    "domain_owner",
    "readonly_source",
    "uat_users",
}


class DomainReplicationError(ValueError):
    """Raised when an implementation package cannot prove a replication gate."""


def _canonical(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")


def _digest(value: Any) -> str:
    return hashlib.sha256(_canonical(value)).hexdigest()


def _mapping(value: Any, field: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise DomainReplicationError(f"{field} must be an object")
    return value


def _bounded_text(value: Any, field: str, limit: int = 256) -> str:
    rendered = str(value or "").strip()
    if not rendered or len(rendered) > limit:
        raise DomainReplicationError(f"{field} must be non-empty and <= {limit}")
    return rendered


def _reject_secrets(value: Any, path: str = "$") -> None:
    if isinstance(value, dict):
        for key, item in value.items():
            if _SECRET.search(str(key)):
                raise DomainReplicationError(
                    f"secret material is not allowed at {path}.{key}"
                )
            _reject_secrets(item, f"{path}.{key}")
    elif isinstance(value, list):
        for index, item in enumerate(value):
            _reject_secrets(item, f"{path}[{index}]")


def _row_key(row: dict[str, str], fields: list[str]) -> tuple[str, ...]:
    values = tuple(str(row.get(field, "")).strip() for field in fields)
    if any(not value for value in values):
        raise DomainReplicationError("source primary key values must be non-empty")
    return values


def _normalized_row(
    row: dict[str, str], *, operation_field: str, cursor_field: str
) -> dict[str, str]:
    return {
        str(key): str(value or "").strip()
        for key, value in sorted(row.items())
        if key not in {operation_field, cursor_field}
    }


def _validate_source_fields(rows: list[dict[str, str]]) -> None:
    for row in rows:
        if not isinstance(row, dict):
            raise DomainReplicationError("source rows must be objects")
        for field in row:
            if _SECRET.search(str(field)):
                raise DomainReplicationError(
                    f"secret-like source field is not allowed: {field}"
                )


def collect_incremental_rows(
    snapshot_rows: list[dict[str, str]],
    delta_rows: list[dict[str, str]],
    source: dict[str, Any],
) -> dict[str, Any]:
    """Apply a controlled snapshot and monotonic delta without exposing rows."""

    source = _mapping(source, "source")
    if source.get("kind") != "controlled_csv":
        raise DomainReplicationError("source.kind must be controlled_csv")
    if source.get("classification") != "desensitized":
        raise DomainReplicationError("source.classification must be desensitized")
    primary_key = source.get("primary_key")
    if not isinstance(primary_key, list) or not primary_key:
        raise DomainReplicationError("source.primary_key must be a non-empty list")
    primary_key = [_bounded_text(item, "source.primary_key") for item in primary_key]
    operation_field = _bounded_text(
        source.get("operation_field", "_operation"), "source.operation_field"
    )
    cursor_field = _bounded_text(
        source.get("cursor_field", "_cursor"), "source.cursor_field"
    )
    required_columns = source.get("required_columns")
    if not isinstance(required_columns, list) or not required_columns:
        raise DomainReplicationError("source.required_columns must be a non-empty list")
    required_columns = {
        _bounded_text(item, "source.required_columns") for item in required_columns
    }
    cursor_before = int(source.get("snapshot_cursor", -1))
    if cursor_before < 0:
        raise DomainReplicationError("source.snapshot_cursor must be non-negative")

    if not snapshot_rows:
        raise DomainReplicationError("controlled snapshot must contain rows")
    if not delta_rows:
        raise DomainReplicationError("controlled delta must contain rows")
    _validate_source_fields(snapshot_rows)
    _validate_source_fields(delta_rows)

    state: dict[tuple[str, ...], dict[str, str]] = {}
    for row in snapshot_rows:
        if not required_columns.issubset(row):
            missing = sorted(required_columns - set(row))
            raise DomainReplicationError(f"snapshot is missing required columns: {missing}")
        key = _row_key(row, primary_key)
        if key in state:
            raise DomainReplicationError("snapshot primary keys must be unique")
        state[key] = _normalized_row(
            row, operation_field=operation_field, cursor_field=cursor_field
        )

    cursors: list[int] = []
    for row in delta_rows:
        if cursor_field not in row or operation_field not in row:
            raise DomainReplicationError("delta cursor and operation fields are required")
        try:
            cursor = int(row[cursor_field])
        except (TypeError, ValueError) as exc:
            raise DomainReplicationError("delta cursor must be an integer") from exc
        if cursor <= cursor_before or (cursors and cursor <= cursors[-1]):
            raise DomainReplicationError("delta cursors must strictly increase")
        cursors.append(cursor)

    def apply_delta() -> dict[str, int]:
        result = {"inserted": 0, "updated": 0, "deleted": 0}
        for row in delta_rows:
            operation = str(row[operation_field]).strip().lower()
            if operation not in {"upsert", "delete"}:
                raise DomainReplicationError(f"unsupported delta operation: {operation}")
            key = _row_key(row, primary_key)
            if operation == "delete":
                if key in state:
                    del state[key]
                    result["deleted"] += 1
                continue
            if not required_columns.issubset(row):
                missing = sorted(required_columns - set(row))
                raise DomainReplicationError(f"delta is missing required columns: {missing}")
            normalized = _normalized_row(
                row, operation_field=operation_field, cursor_field=cursor_field
            )
            if key not in state:
                state[key] = normalized
                result["inserted"] += 1
            elif state[key] != normalized:
                state[key] = normalized
                result["updated"] += 1
        return result

    first = apply_delta()
    first_state = deepcopy(state)
    replay = apply_delta()
    if state != first_state:
        raise DomainReplicationError("delta replay changed final source state")
    replay_changes = sum(replay.values())
    final_rows = [state[key] for key in sorted(state)]
    return {
        "source_kind": "controlled_csv",
        "classification": "desensitized",
        "snapshot_rows": len(snapshot_rows),
        "delta_rows": len(delta_rows),
        **first,
        "final_rows": len(state),
        "replay_changes": replay_changes,
        "cursor_before": cursor_before,
        "cursor_after": cursors[-1],
        "snapshot_digest": _digest(
            sorted(
                (
                    _normalized_row(
                        row,
                        operation_field=operation_field,
                        cursor_field=cursor_field,
                    )
                    for row in snapshot_rows
                ),
                key=_canonical,
            )
        ),
        "delta_digest": _digest(
            [
                {
                    **_normalized_row(
                        row,
                        operation_field=operation_field,
                        cursor_field=cursor_field,
                    ),
                    operation_field: row[operation_field],
                    cursor_field: int(row[cursor_field]),
                }
                for row in delta_rows
            ]
        ),
        "final_digest": _digest(final_rows),
    }


def _positive(metrics: dict[str, Any], field: str, minimum: float = 1) -> float:
    try:
        value = float(metrics.get(field, 0))
    except (TypeError, ValueError) as exc:
        raise DomainReplicationError(f"{field} must be numeric") from exc
    if value < minimum:
        raise DomainReplicationError(f"{field} must be >= {minimum:g}")
    return value


def _validate_stage_metrics(
    stage: str,
    metrics: dict[str, Any],
    *,
    template: dict[str, Any],
    incremental: dict[str, Any],
) -> None:
    if stage == "template_initialization":
        expected = {
            "object_type_count": len(template.get("object_types", [])),
            "rule_count": len(template.get("rules", [])),
            "role_count": len(template.get("responsibility_roles", [])),
            "metric_count": len(template.get("metrics", [])),
        }
        if any(int(metrics.get(key, -1)) != value for key, value in expected.items()):
            raise DomainReplicationError("template initialization counts do not match template")
        _positive(metrics, "template_version")
    elif stage == "incremental_ingestion":
        for key in (
            "snapshot_rows",
            "delta_rows",
            "inserted",
            "updated",
            "deleted",
            "final_rows",
            "replay_changes",
            "cursor_before",
            "cursor_after",
        ):
            if int(metrics.get(key, -1)) != int(incremental[key]):
                raise DomainReplicationError(
                    f"incremental ingestion evidence does not match collected {key}"
                )
    elif stage == "catalog":
        _positive(metrics, "asset_count", 3)
        _positive(metrics, "incremental_change_count")
        if int(metrics.get("cursor_after", -1)) != incremental["cursor_after"]:
            raise DomainReplicationError("catalog cursor does not match ingestion cursor")
    elif stage == "semantics":
        for field in (
            "published_term_count",
            "published_code_set_count",
            "published_metric_count",
            "mapped_field_count",
        ):
            _positive(metrics, field)
    elif stage == "responsibility":
        if float(metrics.get("coverage_percent", 0)) != 100:
            raise DomainReplicationError("responsibility coverage must be 100 percent")
        _positive(metrics, "bound_role_count", 3)
    elif stage == "quality":
        _positive(metrics, "published_rule_count", 5)
        _positive(metrics, "initial_finding_count")
        target = float(metrics.get("target_score", 0))
        final = float(metrics.get("final_score", 0))
        if target <= 0 or final < target:
            raise DomainReplicationError("quality final score must meet target score")
    elif stage == "remediation":
        expected = {
            "issue_status": "closed",
            "task_status": "completed",
            "independent_closer": True,
        }
        if metrics != expected:
            raise DomainReplicationError("remediation issue and task must be independently closed")
    elif stage == "observability":
        _positive(metrics, "slo_count")
        if metrics.get("incident_status") != "closed" or metrics.get("recovered") is not True:
            raise DomainReplicationError("observability incident must recover and close")
    elif stage == "data_product":
        if metrics.get("contract_status") != "active":
            raise DomainReplicationError("data product contract must be active")
        if metrics.get("certificate_status") != "issued":
            raise DomainReplicationError("data product certificate must be issued")
        refs = metrics.get("certificate_evidence_refs")
        if not isinstance(refs, list) or len(refs) < 4:
            raise DomainReplicationError("certificate evidence must bind quality, lineage, rule and workflow")
        required = ("quality://", "lineage://", "rule://", "workflow://")
        if not all(any(str(ref).startswith(prefix) for ref in refs) for prefix in required):
            raise DomainReplicationError("certificate evidence is missing canonical references")
    elif stage == "agent":
        if metrics.get("autonomy_level") not in {"read_only", "suggestion"}:
            raise DomainReplicationError("agent autonomy must be read_only or suggestion")
        if metrics.get("decision") != "authorized":
            raise DomainReplicationError("agent action must be authorized")
        if metrics.get("automatic_execution_allowed") is not False:
            raise DomainReplicationError("agent automatic execution must remain disabled")
        _positive(metrics, "citation_count")
        if metrics.get("cross_domain_denied") is not True:
            raise DomainReplicationError("agent cross-domain request must be denied")


def _validate_manifest(manifest: dict[str, Any], template: dict[str, Any]) -> tuple[str, str]:
    manifest = _mapping(manifest, "manifest")
    _reject_secrets(manifest)
    if int(manifest.get("schema_version", 0)) != 1:
        raise DomainReplicationError("manifest.schema_version must be 1")
    package_code = _bounded_text(manifest.get("package_code"), "package_code", 80)
    domain = _mapping(manifest.get("domain"), "domain")
    domain_code = _bounded_text(domain.get("code"), "domain.code", 64)
    if not _CODE.fullmatch(domain_code):
        raise DomainReplicationError("domain.code must be a stable lowercase code")
    if template.get("template_code") != domain_code:
        raise DomainReplicationError("domain template code does not match package domain")
    if len(template.get("object_types", [])) < 3:
        raise DomainReplicationError("domain template must contain at least three object types")
    if len(template.get("rules", [])) < 5:
        raise DomainReplicationError("domain template must contain at least five rules")

    assessment = _mapping(
        manifest.get("core_change_assessment"), "core_change_assessment"
    )
    device_changes = assessment.get("device_specific_changes")
    if not isinstance(device_changes, list):
        raise DomainReplicationError("device_specific_changes must be a list")
    if device_changes:
        raise DomainReplicationError("device-specific core changes must remain zero")
    generic_extensions = assessment.get("generic_extensions")
    if not isinstance(generic_extensions, list) or not generic_extensions:
        raise DomainReplicationError("generic extension points must be documented")
    if any(
        str(path).startswith(marker)
        for path in generic_extensions
        for marker in _DEVICE_PATH_MARKERS
    ):
        raise DomainReplicationError("generic extensions cannot target device-specific paths")

    reuse = _mapping(manifest.get("third_domain_reuse"), "third_domain_reuse")
    if reuse.get("reusable") is not True:
        raise DomainReplicationError("third-domain reuse must be explicitly enabled")
    placeholders = reuse.get("required_replacements")
    if not isinstance(placeholders, list) or len(placeholders) < 4:
        raise DomainReplicationError("third-domain replacement checklist is incomplete")
    return package_code, domain_code


def evaluate_replication_package(
    manifest: dict[str, Any],
    evidence: dict[str, Any],
    *,
    template: dict[str, Any],
    snapshot_rows: list[dict[str, str]],
    delta_rows: list[dict[str, str]],
) -> dict[str, Any]:
    """Evaluate one domain package and return a deterministic, row-free report."""

    package_code, domain_code = _validate_manifest(manifest, template)
    incremental = collect_incremental_rows(
        snapshot_rows,
        delta_rows,
        _mapping(manifest.get("source"), "source"),
    )
    evidence = _mapping(evidence, "evidence")
    _reject_secrets(evidence)
    if evidence.get("package_code") != package_code:
        raise DomainReplicationError("evidence package code does not match manifest")
    receipts = evidence.get("receipts")
    if not isinstance(receipts, list):
        raise DomainReplicationError("evidence.receipts must be a list")
    by_stage: dict[str, dict[str, Any]] = {}
    for receipt in receipts:
        receipt = _mapping(receipt, "receipt")
        stage = _bounded_text(receipt.get("stage"), "receipt.stage", 80)
        if stage in by_stage:
            raise DomainReplicationError(f"duplicate evidence stage: {stage}")
        by_stage[stage] = receipt
    missing = sorted(set(REQUIRED_STAGES) - set(by_stage))
    extra = sorted(set(by_stage) - set(REQUIRED_STAGES))
    if missing:
        raise DomainReplicationError(f"missing stages: {','.join(missing)}")
    if extra:
        raise DomainReplicationError(f"unsupported stages: {','.join(extra)}")

    stage_reports: dict[str, dict[str, Any]] = {}
    for stage in REQUIRED_STAGES:
        receipt = by_stage[stage]
        if receipt.get("domain_code") != domain_code:
            raise DomainReplicationError(f"{stage} evidence has the wrong domain")
        if receipt.get("status") != "passed":
            raise DomainReplicationError(f"{stage} evidence status must be passed")
        if receipt.get("subsystem") != STAGE_SUBSYSTEMS[stage]:
            raise DomainReplicationError(f"{stage} evidence subsystem is invalid")
        api_refs = receipt.get("api_refs")
        if not isinstance(api_refs, list) or not api_refs:
            raise DomainReplicationError(f"{stage} api_refs are required")
        if any("/device-" in str(ref) for ref in api_refs):
            raise DomainReplicationError("device-specific API evidence is not allowed")
        evidence_refs = receipt.get("evidence_refs")
        if not isinstance(evidence_refs, list) or not evidence_refs:
            raise DomainReplicationError(f"{stage} evidence_refs are required")
        metrics = _mapping(receipt.get("metrics"), f"{stage}.metrics")
        _validate_stage_metrics(
            stage,
            metrics,
            template=template,
            incremental=incremental,
        )
        stage_reports[stage] = {
            "status": "passed",
            "subsystem": STAGE_SUBSYSTEMS[stage],
            "api_refs": sorted(str(ref) for ref in api_refs),
            "evidence_refs": sorted(str(ref) for ref in evidence_refs),
            "metrics": deepcopy(metrics),
        }

    bindings = manifest.get("enterprise_bindings")
    if not isinstance(bindings, list) or not bindings:
        raise DomainReplicationError("enterprise binding checklist is required")
    binding_statuses: dict[str, str] = {}
    for item in bindings:
        item = _mapping(item, "enterprise_bindings.item")
        kind = _bounded_text(item.get("kind"), "enterprise_bindings.kind")
        status = _bounded_text(item.get("status"), "enterprise_bindings.status")
        if status not in {"bound", "unbound"}:
            raise DomainReplicationError("enterprise binding status must be bound or unbound")
        if kind in binding_statuses:
            raise DomainReplicationError(f"duplicate enterprise binding: {kind}")
        binding_statuses[kind] = status
    missing_bindings = sorted(_REQUIRED_ENTERPRISE_BINDINGS - set(binding_statuses))
    if missing_bindings:
        raise DomainReplicationError(
            f"missing enterprise bindings: {','.join(missing_bindings)}"
        )
    unbound = sorted(
        kind for kind, status in binding_statuses.items() if status == "unbound"
    )
    assessment = manifest["core_change_assessment"]
    report: dict[str, Any] = {
        "schema_version": 1,
        "package_code": package_code,
        "status": "passed",
        "domain": {
            "code": domain_code,
            "name": _bounded_text(manifest["domain"].get("name"), "domain.name"),
            "template_version": int(
                stage_reports["template_initialization"]["metrics"]["template_version"]
            ),
        },
        "incremental_collection": incremental,
        "stages": stage_reports,
        "device_specific_core_changes": 0,
        "generic_extension_count": len(assessment["generic_extensions"]),
        "third_domain_reusable": True,
        "enterprise_uat": {
            "status": "blocked_external" if unbound else "ready",
            "unbound_requirements": unbound,
        },
    }
    report["report_sha256"] = _digest(report)
    return report