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