"""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 import unicodedata from copy import deepcopy from datetime import date from typing import Any from app.core.governance.domain_replication_contract_registry import verify_registry 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", } _EXECUTION_ACTIONS = ("dry_run", "execute") _RECOVERY_ACTIONS = frozenset({"diff", "rollback"}) _EXECUTION_REF = re.compile( r"^[a-z][a-z0-9-]{1,31}://([a-z][a-z0-9_]{1,63})/[a-zA-Z0-9_./:-]{1,240}$" ) _HTTP_URL = re.compile(r"https?://", re.IGNORECASE) _RAW_ROW_KEY = re.compile(r"(?:raw[_-]?rows?|source[_-]?rows?)", re.IGNORECASE) _SHA256 = re.compile(r"^[0-9a-f]{64}$") _IDENTIFIER = re.compile(r"^[a-z][a-z0-9_-]{1,79}$") _CONFUSABLES = str.maketrans({ "а": "a", "е": "e", "і": "i", "о": "o", "р": "p", "с": "c", "х": "x", "у": "y", "к": "k", "м": "m", "т": "t", "ѕ": "s", }) _SENSITIVE_KEY_MARKERS = ( "password", "passwd", "secret", "token", "credential", "authorization", "apikey", "privatekey", ) _RAW_SOURCE_KEYS = {"rawrows", "sourcerows", "rows"} _SQL = re.compile(r"\b(?:select|insert|update|delete|drop|alter|create|grant|revoke)\b", re.IGNORECASE) _API_ALLOWLIST = { "template_initialization": ("/api/meta/domain-templates",), "incremental_ingestion": ("/api/meta/active-metadata",), "catalog": ("/api/meta/active-metadata",), "semantics": ("/api/development/v1/semantic-assets",), "responsibility": ("/api/system/responsibilities",), "quality": ("/api/rules/quality-operations",), "remediation": ("/api/system/work-center",), "observability": ("/api/datafactory/observability",), "data_product": ("/api/dataservice/governance/products",), "agent": ("/api/knowledge/agents",), } _EVIDENCE_PREFIXES = { "template_initialization": "domain-template://", "incremental_ingestion": "ingestion://", "catalog": "catalog-run://", "semantics": "semantic-release://", "responsibility": "responsibility://", "quality": "quality://", "remediation": "workflow://", "observability": "incident://", "data_product": "product://", "agent": "agent-request://", } _CONTRACT_REQUIREMENTS = { "incremental_ingestion": "p3_wp03_connector_run", "catalog": "p3_wp03_catalog_lineage", "observability": "p3_wp04_edge_health", } _METRIC_FIELDS = { "template_initialization": {"template_version", "object_type_count", "rule_count", "role_count", "metric_count"}, "incremental_ingestion": {"snapshot_rows", "delta_rows", "inserted", "updated", "deleted", "final_rows", "replay_changes", "cursor_before", "cursor_after"}, "catalog": {"asset_count", "incremental_change_count", "cursor_after"}, "semantics": {"published_term_count", "published_code_set_count", "published_metric_count", "mapped_field_count"}, "responsibility": {"coverage_percent", "bound_role_count"}, "quality": {"published_rule_count", "initial_finding_count", "target_score", "final_score"}, "remediation": {"issue_status", "task_status", "independent_closer"}, "observability": {"slo_count", "incident_status", "recovered"}, "data_product": {"contract_status", "certificate_status", "certificate_evidence_refs"}, "agent": {"autonomy_level", "decision", "automatic_execution_allowed", "citation_count", "cross_domain_denied"}, } 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 _security_key(value: Any) -> str: return re.sub( r"[^a-z0-9]", "", unicodedata.normalize("NFKC", str(value)).casefold().translate(_CONFUSABLES) ) def _reject_secrets(value: Any, path: str = "$", depth: int = 0) -> None: if depth > 12: raise DomainReplicationError(f"maximum package depth exceeded at {path}") if isinstance(value, dict): if len(value) > 64: raise DomainReplicationError(f"maximum object properties exceeded at {path}") for key, item in value.items(): normalized_key = _security_key(key) if normalized_key in _RAW_SOURCE_KEYS: raise DomainReplicationError( f"raw source rows are not allowed at {path}.{key}" ) if _SECRET.search(normalized_key) or any( marker in normalized_key for marker in _SENSITIVE_KEY_MARKERS ): raise DomainReplicationError( f"sensitive material is not allowed at {path}.{key}" ) _reject_secrets(item, f"{path}.{key}", depth + 1) elif isinstance(value, list): if len(value) > 500: raise DomainReplicationError(f"maximum collection items exceeded at {path}") for index, item in enumerate(value): _reject_secrets(item, f"{path}[{index}]", depth + 1) elif isinstance(value, str): if len(value) > 4096: raise DomainReplicationError(f"maximum string length exceeded at {path}") normalized = unicodedata.normalize("NFKC", value) if _HTTP_URL.search(normalized): raise DomainReplicationError(f"URL is not allowed at {path}") if _SQL.search(normalized): raise DomainReplicationError(f"SQL is not allowed at {path}") def _closed_mapping(value: Any, field: str, allowed: set[str]) -> dict[str, Any]: value = _mapping(value, field) unknown = sorted(set(value) - allowed) if unknown: normalized_unknown = [_security_key(key) for key in unknown] if any(key in _RAW_SOURCE_KEYS for key in normalized_unknown): raise DomainReplicationError(f"raw source rows are not allowed at {field}") if any( _SECRET.search(key) or any(marker in key for marker in _SENSITIVE_KEY_MARKERS) for key in normalized_unknown ): raise DomainReplicationError(f"sensitive material is not allowed at {field}") raise DomainReplicationError(f"{field} has unsupported fields: {unknown}") return value def _closed_required(value: Any, field: str, required: set[str], allowed: set[str]) -> dict[str, Any]: value = _closed_mapping(value, field, allowed) missing = sorted(required - set(value)) if missing: raise DomainReplicationError(f"{field} is missing required fields: {missing}") return value def _string(value: Any, field: str, limit: int = 256) -> str: if not isinstance(value, str): raise DomainReplicationError(f"{field} must be a string") return _bounded_text(value, field, limit) def _digest_string(value: Any, field: str) -> str: value = _string(value, field, 64) if not _SHA256.fullmatch(value): raise DomainReplicationError(f"{field} must be a lowercase SHA-256 digest") return value def _bounded_list(value: Any, field: str, *, minimum: int = 1, maximum: int = 100) -> list[Any]: if not isinstance(value, list) or not minimum <= len(value) <= maximum: raise DomainReplicationError(f"{field} must contain {minimum}..{maximum} items") return value def _execution_ref(value: Any, field: str) -> tuple[str, str]: rendered = _bounded_text(value, field, 320) if _HTTP_URL.search(rendered): raise DomainReplicationError(f"{field} must not contain a URL") matched = _EXECUTION_REF.fullmatch(rendered) if not matched: raise DomainReplicationError(f"{field} must be a canonical evidence reference") return rendered, matched.group(1) def _reject_execution_sensitive(value: Any, path: str = "execution_evidence") -> None: if isinstance(value, dict): for key, item in value.items(): if _RAW_ROW_KEY.search(str(key)): raise DomainReplicationError(f"raw source rows are not allowed at {path}.{key}") if _HTTP_URL.search(str(item)): raise DomainReplicationError(f"URL is not allowed at {path}.{key}") _reject_execution_sensitive(item, f"{path}.{key}") elif isinstance(value, list): for index, item in enumerate(value): _reject_execution_sensitive(item, f"{path}[{index}]") def _validate_execute_state(action: dict[str, Any], stage: str) -> dict[str, Any]: """Validate the v2 receipt that owns a governed state transition.""" required = { "action", "status", "write_count", "evidence_ref", "audit_ref", "operation_uid", "idempotency_key", "request_digest", "attempt", "lease_fence", "pre_state_digest", "post_state_digest", "result_digest", } action = _closed_required(action, f"{stage}.execute", required, required) for field in ("operation_uid", "idempotency_key"): if not _IDENTIFIER.fullmatch(_string(action[field], f"{stage}.execute.{field}", 80)): raise DomainReplicationError(f"{stage}.execute.{field} is invalid") for field in ("request_digest", "pre_state_digest", "post_state_digest", "result_digest"): _digest_string(action[field], f"{stage}.execute.{field}") if action["pre_state_digest"] == action["post_state_digest"]: raise DomainReplicationError(f"{stage}.execute state digests must differ") if isinstance(action["attempt"], bool) or not isinstance(action["attempt"], int) or not 1 <= action["attempt"] <= 5: raise DomainReplicationError(f"{stage}.execute.attempt must be 1..5") if isinstance(action["lease_fence"], bool) or not isinstance(action["lease_fence"], int) or action["lease_fence"] < 1: raise DomainReplicationError(f"{stage}.execute.lease_fence must be positive") return action def _validate_rollback(action: dict[str, Any], stage: str, *, strict_state_machine: bool) -> dict[str, Any]: required = { "action", "status", "write_count", "evidence_ref", "audit_ref", "operation", "idempotency_key", "request_digest", "attempt", "lease_fence", "from_digest", "to_digest", "replay_status", } if strict_state_machine: required = required - {"replay_status"} | {"operation_uid", "result_digest"} action = _closed_required(action, f"{stage}.rollback", required, required) if action["operation"] != "restore": raise DomainReplicationError(f"{stage}.rollback.operation must be restore") for field in ("operation_uid", "idempotency_key") if strict_state_machine else ("idempotency_key",): if not _IDENTIFIER.fullmatch(_string(action[field], f"{stage}.rollback.{field}", 80)): raise DomainReplicationError(f"{stage}.rollback.{field} is invalid") digest_fields = ("request_digest", "from_digest", "to_digest", "result_digest") if strict_state_machine else ("request_digest", "from_digest", "to_digest") for key in digest_fields: _digest_string(action[key], f"{stage}.rollback.{key}") if action["from_digest"] == action["to_digest"]: raise DomainReplicationError(f"{stage}.rollback digests must differ") if isinstance(action["attempt"], bool) or not isinstance(action["attempt"], int) or not 1 <= action["attempt"] <= 5: raise DomainReplicationError(f"{stage}.rollback.attempt must be 1..5") if isinstance(action["lease_fence"], bool) or not isinstance(action["lease_fence"], int) or action["lease_fence"] < 1: raise DomainReplicationError(f"{stage}.rollback.lease_fence must be positive") if not strict_state_machine and action["replay_status"] != "exact_replay": raise DomainReplicationError(f"{stage}.rollback replay must be exact_replay") return action def _validate_replay_receipts(value: Any, rollbacks: dict[str, dict[str, Any]]) -> None: """Require a second receipt, not a rollback's self-declared replay label.""" required = { "stage", "operation_uid", "idempotency_key", "request_digest", "attempt", "lease_fence", "from_digest", "to_digest", "result_digest", "status", } receipts = _bounded_list(value, "execution_evidence.replay_receipts", minimum=len(rollbacks), maximum=len(rollbacks)) by_stage: dict[str, dict[str, Any]] = {} for item in receipts: item = _closed_required(item, "execution_evidence.replay_receipt", required, required) stage = _string(item["stage"], "execution_evidence.replay_receipt.stage", 80) if stage not in rollbacks or stage in by_stage: raise DomainReplicationError("replay receipt must bind exactly one rollback stage") if item["status"] != "exact_replay": raise DomainReplicationError("replay receipt status must be exact_replay") for field in ("operation_uid", "idempotency_key"): if not _IDENTIFIER.fullmatch(_string(item[field], f"replay_receipt.{field}", 80)): raise DomainReplicationError(f"replay receipt {field} is invalid") for field in ("request_digest", "from_digest", "to_digest", "result_digest"): _digest_string(item[field], f"replay_receipt.{field}") rollback = rollbacks[stage] for field in required - {"stage", "status"}: if item[field] != rollback[field]: raise DomainReplicationError(f"replay receipt does not match rollback {field}") by_stage[stage] = item if set(by_stage) != set(rollbacks): raise DomainReplicationError("replay receipt coverage is incomplete") def validate_execution_evidence( execution_evidence: dict[str, Any], *, domain_code: str | None = None ) -> dict[str, dict[str, Any]]: """Validate the ordered, row-free execution chain for every governance stage.""" initial = _mapping(execution_evidence, "execution_evidence") schema_version = initial.get("schema_version") if schema_version not in {1, 2}: raise DomainReplicationError("execution_evidence.schema_version must be 1 or 2") strict_state_machine = schema_version == 2 execution_evidence = _closed_mapping( execution_evidence, "execution_evidence", {"schema_version", "stages", "replay_receipts"} if strict_state_machine else {"schema_version", "stages"}, ) _reject_secrets(execution_evidence, "execution_evidence") _reject_execution_sensitive(execution_evidence) receipts = execution_evidence.get("stages") if not isinstance(receipts, list) or len(receipts) != len(REQUIRED_STAGES): raise DomainReplicationError("execution_evidence must contain every required stage") summaries: dict[str, dict[str, Any]] = {} rollbacks: dict[str, dict[str, Any]] = {} for expected_stage, receipt in zip(REQUIRED_STAGES, receipts, strict=True): receipt = _closed_mapping(receipt, "execution_evidence.stage", {"stage", "actions"}) stage = _bounded_text(receipt.get("stage"), "execution_evidence.stage.stage", 80) if stage != expected_stage: raise DomainReplicationError("execution evidence stages must be ordered") actions = receipt.get("actions") if not isinstance(actions, list) or len(actions) != 3: raise DomainReplicationError(f"{stage} must contain dry_run, execute and diff or rollback") normalized_actions: list[dict[str, Any]] = [] domains: set[str] = set() for index, action in enumerate(actions): action = _mapping(action, f"{stage}.actions[{index}]") action_name = _bounded_text(action.get("action"), f"{stage}.action", 32) expected_action = ( _EXECUTION_ACTIONS[index] if index < 2 else None ) if expected_action and action_name != expected_action: raise DomainReplicationError(f"{stage} actions must be ordered dry_run then execute") if index == 2 and action_name not in _RECOVERY_ACTIONS: raise DomainReplicationError(f"{stage} recovery action must be diff or rollback") allowed = {"action", "status", "write_count", "evidence_ref", "audit_ref"} if action_name == "rollback": action = _validate_rollback(action, stage, strict_state_machine=strict_state_machine) elif action_name == "execute" and strict_state_machine: action = _validate_execute_state(action, stage) else: action = _closed_required(action, f"{stage}.actions[{index}]", allowed, allowed) if action.get("status") != "passed": raise DomainReplicationError(f"{stage}.{action_name} status must be passed") write_count = action.get("write_count") if isinstance(write_count, bool) or not isinstance(write_count, int): raise DomainReplicationError(f"{stage}.{action_name}.write_count must be an integer") if write_count < 0 or write_count > 1_000_000: raise DomainReplicationError(f"{stage}.{action_name}.write_count is out of range") if action_name == "dry_run" and write_count != 0: raise DomainReplicationError(f"{stage}.dry_run must record zero writes") evidence_ref, evidence_domain = _execution_ref( action.get("evidence_ref"), f"{stage}.{action_name}.evidence_ref" ) audit_ref, audit_domain = _execution_ref( action.get("audit_ref"), f"{stage}.{action_name}.audit_ref" ) domains.update({evidence_domain, audit_domain}) normalized_actions.append( { "action": action_name, "write_count": write_count, "evidence_ref": evidence_ref, "audit_ref": audit_ref, } ) if action_name == "rollback" and strict_state_machine: execute = actions[1] for field in ("operation_uid", "idempotency_key", "request_digest", "attempt", "lease_fence"): if action[field] != execute[field]: raise DomainReplicationError(f"{stage}.rollback must match execute {field}") if action["from_digest"] != execute["post_state_digest"]: raise DomainReplicationError(f"{stage}.rollback.from_digest must match execute post_state_digest") if action["to_digest"] != execute["pre_state_digest"]: raise DomainReplicationError(f"{stage}.rollback.to_digest must match execute pre_state_digest") rollbacks[stage] = action if len(domains) != 1 or (domain_code is not None and domains != {domain_code}): raise DomainReplicationError(f"{stage} execution evidence contains a cross-domain reference") summaries[stage] = { "actions": [item["action"] for item in normalized_actions], "dry_run_write_count": normalized_actions[0]["write_count"], "execute_write_count": normalized_actions[1]["write_count"], "recovery_strategy": normalized_actions[2]["action"], "recovery_write_count": normalized_actions[2]["write_count"], "audit_ref_count": len(normalized_actions), "evidence_digest": _digest(normalized_actions), } if strict_state_machine: _validate_replay_receipts(execution_evidence.get("replay_receipts"), rollbacks) return summaries 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: normalized = _security_key(field) if normalized in _RAW_SOURCE_KEYS or _SECRET.search(normalized) or any( marker in normalized for marker in _SENSITIVE_KEY_MARKERS ): 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_seed_data(seed_data: Any, *, schema_version: int) -> None: payload_key = "records" if schema_version == 1 else "items" record_fields = { "source_contract": {"id", "type", "binding_status"}, "term_and_code_set": {"id", "name"}, "workflow_contract": {"id", "name"}, } for item in _bounded_list(seed_data, "template.seed_data", minimum=0): item = _closed_required( item, "template.seed_data.item", {"kind", payload_key}, {"kind", payload_key}, ) kind = _string(item["kind"], "template.seed_data.kind", 80) if kind not in record_fields: raise DomainReplicationError("template.seed_data.kind is unsupported") for record in _bounded_list(item[payload_key], f"template.seed_data.{payload_key}", minimum=1): record = _closed_required( record, "template.seed_data.record", record_fields[kind], record_fields[kind], ) for key in record_fields[kind]: _string(record[key], f"template.seed_data.record.{key}", 200) def _validate_template(template: dict[str, Any], *, schema_version: int) -> dict[str, Any]: required = { "template_code", "name", "description", "lifecycle_status", "object_types", "responsibility_roles", "rules", "metrics", "seed_data", } _reject_secrets(template, "template") template = _closed_required(template, "template", required, required) _string(template["template_code"], "template.template_code", 64) _string(template["name"], "template.name", 200) _string(template["description"], "template.description", 500) if template["lifecycle_status"] not in {"draft", "active", "retired"}: raise DomainReplicationError("template.lifecycle_status is invalid") for item in _bounded_list(template["object_types"], "template.object_types", minimum=3): item = _closed_required( item, "template.object_types.item", {"type_code", "name", "stable_uid_prefix", "source_identity_fields", "fields"}, {"type_code", "name", "description", "stable_uid_prefix", "source_identity_fields", "fields"}, ) for key in ("type_code", "name", "stable_uid_prefix"): _string(item[key], f"template.object_types.{key}", 200) for identity in _bounded_list(item["source_identity_fields"], "template.source_identity_fields"): _string(identity, "template.source_identity_fields.item", 64) for field in _bounded_list(item["fields"], "template.fields"): field = _closed_required( field, "template.fields.item", {"code", "name", "type", "required"}, {"code", "name", "type", "required", "description"}, ) if not isinstance(field["required"], bool): raise DomainReplicationError("template.fields.required must be boolean") for key in ("code", "name", "type"): _string(field[key], f"template.fields.{key}", 200) for collection, required_fields in ( ("responsibility_roles", {"code", "name", "raci_role"}), ("rules", {"code", "name", "dimension", "target_object"}), ("metrics", {"code", "name", "unit"}), ): for item in _bounded_list(template[collection], f"template.{collection}"): item = _closed_required(item, f"template.{collection}.item", required_fields, required_fields) for key in required_fields: _string(item[key], f"template.{collection}.{key}", 200) _validate_seed_data(template["seed_data"], schema_version=schema_version) return template def _validate_contract_refs(stage: str, value: Any) -> list[dict[str, Any]]: if stage not in _CONTRACT_REQUIREMENTS: if value is not None: raise DomainReplicationError(f"{stage} does not accept contract_refs") return [] refs = _bounded_list(value, f"{stage}.contract_refs", minimum=1, maximum=1) return [validate_contract_reference(refs[0], expected_contract_id=_CONTRACT_REQUIREMENTS[stage])] def validate_contract_reference( reference: dict[str, Any], *, expected_contract_id: str ) -> dict[str, Any]: """Resolve one v2 receipt only against the code-side trusted registry.""" required = { "contract_id", "contract_version", "schema_digest", "operation_uid", "run_id", "operation", "attempt", "lease_fence", "request_digest", "result_digest", "status", } reference = _closed_required(reference, "contract_ref", required, required) try: registry = verify_registry() except ValueError as exc: raise DomainReplicationError("trusted contract registry integrity check failed") from exc if registry["trust_level"] != "ENGINEERING_EVIDENCE_ONLY": raise DomainReplicationError("trusted contract registry has an invalid trust level") if date.fromisoformat(registry["expires_on"]) < date.today(): raise DomainReplicationError("trusted contract registry is expired") contract_id = _string(reference["contract_id"], "contract_ref.contract_id", 80) if contract_id != expected_contract_id: raise DomainReplicationError("cross-contract reference was rejected") contract = next( (item for item in registry["contracts"] if item["contract_id"] == contract_id), None ) if contract is None: raise DomainReplicationError("contract is absent from the trusted registry") if reference["contract_version"] != contract["contract_version"]: raise DomainReplicationError("contract version does not match the trusted registry") if reference["schema_digest"] != contract["schema_digest"]: raise DomainReplicationError("contract schema digest does not match the trusted registry") if reference["operation"] not in contract["allowed_operations"]: raise DomainReplicationError("contract operation is not allowed") run = next( ( item for item in contract["runs"] if item["operation_uid"] == reference["operation_uid"] and item["run_id"] == reference["run_id"] ), None, ) if run is None: raise DomainReplicationError("run is absent from the trusted registry") for field in ( "operation_uid", "run_id", "operation", "attempt", "lease_fence", "request_digest", "result_digest", "status", ): if reference[field] != run[field]: raise DomainReplicationError(f"contract registry mismatch for {field}") for field in ("schema_digest", "request_digest", "result_digest"): _digest_string(reference[field], f"contract_ref.{field}") return deepcopy(reference) def _validate_terminal_evidence( evidence: dict[str, Any], package_code: str, domain_code: str, template: dict[str, Any], incremental: dict[str, Any], *, schema_version: int ) -> dict[str, dict[str, Any]]: _reject_secrets(evidence, "evidence") evidence = _closed_required(evidence, "evidence", {"schema_version", "package_code", "receipts"}, {"schema_version", "package_code", "receipts"}) if evidence["schema_version"] != 1 or evidence["package_code"] != package_code: raise DomainReplicationError("evidence package identity is invalid") receipts = _bounded_list(evidence["receipts"], "evidence.receipts", minimum=1, maximum=len(REQUIRED_STAGES)) by_stage: dict[str, dict[str, Any]] = {} for receipt in receipts: receipt = _mapping(receipt, "receipt") stage = _string(receipt.get("stage"), "receipt.stage", 80) required = {"stage", "domain_code", "status", "subsystem", "api_refs", "evidence_refs", "metrics"} allowed = set(required) if schema_version == 2 and stage in _CONTRACT_REQUIREMENTS: required.add("contract_refs") allowed.add("contract_refs") receipt = _closed_required(receipt, "receipt", required, allowed) 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)}") reports: dict[str, dict[str, Any]] = {} for stage in REQUIRED_STAGES: receipt = by_stage[stage] if receipt["domain_code"] != domain_code: raise DomainReplicationError(f"{stage} evidence has the wrong domain") if receipt["status"] != "passed": raise DomainReplicationError(f"{stage} evidence status must be passed") if receipt["subsystem"] != STAGE_SUBSYSTEMS[stage]: raise DomainReplicationError(f"{stage} evidence subsystem is invalid") api_refs = _bounded_list(receipt["api_refs"], f"{stage}.api_refs", maximum=1) if any("/device-" in str(ref) for ref in api_refs): raise DomainReplicationError("device-specific API evidence is not allowed") if tuple(api_refs) != _API_ALLOWLIST[stage]: raise DomainReplicationError(f"{stage} must use its allowlisted API") evidence_refs = _bounded_list(receipt["evidence_refs"], f"{stage}.evidence_refs", maximum=8) prefix = _EVIDENCE_PREFIXES[stage] if any(not isinstance(ref, str) or not ref.startswith(prefix + domain_code + "/") for ref in evidence_refs): raise DomainReplicationError(f"{stage} evidence_refs are not canonical") metrics = _closed_required(receipt["metrics"], f"{stage}.metrics", _METRIC_FIELDS[stage], _METRIC_FIELDS[stage]) _validate_stage_metrics(stage, metrics, template=template, incremental=incremental) contracts = ( _validate_contract_refs(stage, receipt.get("contract_refs")) if schema_version == 2 else [] ) reports[stage] = {"status": "passed", "subsystem": STAGE_SUBSYSTEMS[stage], "api_refs": list(api_refs), "evidence_refs": sorted(evidence_refs), "metrics": deepcopy(metrics)} if contracts: reports[stage]["contract_refs"] = contracts return reports def _validate_manifest( manifest: dict[str, Any], template: dict[str, Any] ) -> tuple[str, str, int]: required = { "schema_version", "package_code", "domain", "files", "source", "core_change_assessment", "enterprise_bindings", "third_domain_reuse", } _reject_secrets(manifest) manifest = _closed_required(manifest, "manifest", required, required) if isinstance(manifest["schema_version"], bool) or not isinstance(manifest["schema_version"], int): raise DomainReplicationError("manifest.schema_version must be an integer") schema_version = manifest["schema_version"] if schema_version not in {1, 2}: raise DomainReplicationError("manifest.schema_version must be 1 or 2") package_code = _string(manifest["package_code"], "package_code", 80) domain = _closed_required(manifest["domain"], "manifest.domain", {"code", "name"}, {"code", "name"}) domain_code = _string(domain["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") files = _mapping(manifest["files"], "manifest.files") expected_file_keys = {"template", "evidence", "snapshot", "delta", "acceptance_report"} if schema_version == 2: expected_file_keys.add("execution_evidence") if set(files) != expected_file_keys: raise DomainReplicationError("manifest.files has unsupported or missing bindings") for key, binding in files.items(): binding = _closed_required(binding, f"manifest.files.{key}", {"path", "sha256"}, {"path", "sha256"}) _string(binding["path"], f"manifest.files.{key}.path", 240) _digest_string(binding["sha256"], f"manifest.files.{key}.sha256") source = _closed_required( manifest["source"], "manifest.source", {"kind", "classification", "primary_key", "required_columns", "operation_field", "cursor_field", "snapshot_cursor"}, {"kind", "classification", "primary_key", "required_columns", "operation_field", "cursor_field", "snapshot_cursor"}, ) if source["kind"] != "controlled_csv" or source["classification"] != "desensitized": raise DomainReplicationError("manifest source is invalid") for field in ("primary_key", "required_columns"): for value in _bounded_list(source[field], f"manifest.source.{field}"): _string(value, f"manifest.source.{field}.item", 80) for field in ("operation_field", "cursor_field"): _string(source[field], f"manifest.source.{field}", 80) if isinstance(source["snapshot_cursor"], bool) or not isinstance(source["snapshot_cursor"], int) or source["snapshot_cursor"] < 0: raise DomainReplicationError("manifest.source.snapshot_cursor is invalid") assessment = _closed_required( manifest["core_change_assessment"], "core_change_assessment", {"baseline_commit", "device_specific_changes", "generic_extensions", "extension_points_used"}, {"baseline_commit", "device_specific_changes", "generic_extensions", "extension_points_used"}, ) 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 = _bounded_list(assessment["generic_extensions"], "generic_extensions") 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 = _closed_required( manifest["third_domain_reuse"], "third_domain_reuse", {"reusable", "required_replacements"}, {"reusable", "required_replacements"}, ) if reuse.get("reusable") is not True: raise DomainReplicationError("third-domain reuse must be explicitly enabled") placeholders = reuse["required_replacements"] if not isinstance(placeholders, list) or len(placeholders) < 4: raise DomainReplicationError("third-domain replacement checklist is incomplete") return package_code, domain_code, schema_version 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]], execution_evidence: dict[str, Any] | None = None, ) -> dict[str, Any]: """Evaluate one domain package and return a deterministic, row-free report.""" manifest_schema_version = _mapping(manifest, "manifest").get("schema_version") if isinstance(manifest_schema_version, bool) or manifest_schema_version not in {1, 2}: raise DomainReplicationError("manifest.schema_version must be 1 or 2") template = _validate_template(template, schema_version=manifest_schema_version) package_code, domain_code, schema_version = _validate_manifest(manifest, template) incremental = collect_incremental_rows( snapshot_rows, delta_rows, _mapping(manifest.get("source"), "source"), ) stage_reports = _validate_terminal_evidence( evidence, package_code, domain_code, template, incremental, schema_version=schema_version ) bindings = _bounded_list(manifest["enterprise_bindings"], "enterprise_bindings", minimum=1, maximum=4) binding_statuses: dict[str, str] = {} for item in bindings: item = _closed_required(item, "enterprise_bindings.item", {"kind", "status"}, {"kind", "status"}) kind = _string(item["kind"], "enterprise_bindings.kind") status = _string(item["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": schema_version, "package_code": package_code, "status": "passed", "domain": { "code": domain_code, "name": _string(manifest["domain"]["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, }, } if schema_version == 2: if execution_evidence is None: raise DomainReplicationError("schema-v2 package requires execution evidence") report["execution_evidence"] = validate_execution_evidence( execution_evidence, domain_code=domain_code ) elif execution_evidence is not None: raise DomainReplicationError("schema-v1 package cannot include execution evidence") report["report_sha256"] = _digest(report) return report