"""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