DataOps-platform/app/core/data_rules/contracts.py

"""Closed, deterministic contracts for AI-authored data processing assets."""

from __future__ import annotations

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

from app.core.common.identifiers import ensure_governance_uid
from app.core.data_rules.expressions import parse_expression

SCHEMA_VERSION = "1.0"
RULE_SCHEMA_VERSION = "2.0"
LEGACY_RULE_SCHEMA_VERSION = "1.0"
RULE_OPS = {
    "aggregate",
    "assert",
    "cast",
    "deduplicate",
    "derive",
    "fill_null",
    "filter",
    "lookup_join",
    "map_values",
    "mask",
    "normalize_text",
    "regex_replace",
}
COMPONENT_TYPES = {"standard.enforce", "rule.apply", "quality.check"}
STAGES = {"extract", "normalize", "transform", "quality_gate", "write", "publish"}
SEVERITIES = {"info", "warning", "error", "critical"}
FAILURE_ACTIONS = {"reject", "quarantine", "warn", "fail"}
NULL_POLICIES = {"explicit", "preserve", "reject"}
IDEMPOTENCY_STRATEGIES = {
    "partition_replace",
    "upsert",
    "deduplication_key",
}
SECRET_KEY_NAMES = {
    "apikey",
    "authorization",
    "connectionstring",
    "credential",
    "credentials",
    "dsn",
    "password",
    "secret",
    "token",
}

RULE_ROOT_KEYS = {
    "schema_version",
    "rule_uid",
    "name",
    "description",
    "input_schema_ref",
    "output_schema_ref",
    "steps",
    "null_policy",
    "timezone",
}
RULE_STEP_KEYS = {
    "id",
    "op",
    "column",
    "target",
    "to",
    "expression",
    "expression_text",
    "expression_ast",
    "pattern",
    "replacement",
    "value",
    "trim",
    "lowercase",
    "uppercase",
    "on_error",
    "on_failure",
    "severity",
    "keys",
    "keep",
    "order_by",
    "mapping",
    "group_by",
    "aggregations",
    "lookup",
    "policy",
}
STANDARD_ROOT_KEYS = {
    "schema_version",
    "standard_uid",
    "name",
    "description",
    "scope",
    "clauses",
}
STANDARD_SCOPE_KEYS = {"object_type", "schema_ref", "business_domain_uid"}
STANDARD_CLAUSE_KEYS = {
    "id",
    "description",
    "severity",
    "rule_version_id",
    "exception_policy",
}
DATAFLOW_ROOT_KEYS = {
    "schema_version",
    "dataflow_uid",
    "name",
    "description",
    "input_schema_refs",
    "output_schema_ref",
    "components",
    "parameters",
}
COMPONENT_KEYS = {
    "id",
    "type",
    "rule_version_id",
    "standard_version_id",
    "stage",
    "order",
    "idempotency",
}
IDEMPOTENCY_KEYS = {"strategy", "key"}
CANDIDATE_KEYS = {
    "schema_version",
    "candidate_type",
    "rule_spec",
    "standard_spec",
    "assumptions",
    "ambiguities",
    "confidence",
    "explanation",
}


RULE_SPEC_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://dataops.local/schemas/rule-spec-2.0.json",
    "title": "DataOps RuleSpec",
    "type": "object",
    "additionalProperties": False,
    "required": sorted(RULE_ROOT_KEYS - {"description"}),
    "properties": {
        "schema_version": {"const": RULE_SCHEMA_VERSION},
        "rule_uid": {"type": "string", "format": "uuid"},
        "name": {"type": "string", "minLength": 1, "maxLength": 200},
        "description": {"type": "string", "maxLength": 2000},
        "input_schema_ref": {"type": "string", "minLength": 1, "maxLength": 500},
        "output_schema_ref": {"type": "string", "minLength": 1, "maxLength": 500},
        "steps": {
            "type": "array",
            "minItems": 1,
            "maxItems": 200,
            "items": {
                "type": "object",
                "additionalProperties": False,
                "required": ["id", "op"],
                "properties": {
                    "id": {
                        "type": "string",
                        "pattern": "^[A-Za-z][A-Za-z0-9_-]{0,99}$",
                    },
                    "op": {"enum": sorted(RULE_OPS)},
                    "column": {"type": "string", "minLength": 1},
                    "target": {"type": "string", "minLength": 1},
                    "to": {"type": "string", "minLength": 1},
                    "expression": {"type": "string", "minLength": 1},
                    "pattern": {"type": "string"},
                    "replacement": {"type": "string"},
                    "value": {},
                    "trim": {"type": "boolean"},
                    "lowercase": {"type": "boolean"},
                    "uppercase": {"type": "boolean"},
                    "on_error": {"enum": sorted(FAILURE_ACTIONS)},
                    "on_failure": {"enum": sorted(FAILURE_ACTIONS)},
                    "severity": {"enum": sorted(SEVERITIES)},
                    "keys": {
                        "type": "array",
                        "items": {"type": "string", "minLength": 1},
                    },
                    "keep": {"enum": ["first", "last"]},
                    "order_by": {
                        "type": "array",
                        "items": {"type": "string", "minLength": 1},
                    },
                    "mapping": {"type": "object"},
                    "group_by": {
                        "type": "array",
                        "items": {"type": "string", "minLength": 1},
                    },
                    "aggregations": {"type": "object"},
                    "lookup": {"type": "object"},
                    "policy": {"type": "string", "minLength": 1},
                },
            },
        },
        "null_policy": {"enum": sorted(NULL_POLICIES)},
        "timezone": {"type": "string", "minLength": 1, "maxLength": 100},
    },
}

STANDARD_SPEC_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://dataops.local/schemas/data-standard-spec-1.0.json",
    "title": "DataOps DataStandardSpec",
    "type": "object",
    "additionalProperties": False,
    "required": ["schema_version", "standard_uid", "name", "scope", "clauses"],
}

DATAFLOW_SPEC_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://dataops.local/schemas/dataflow-production-line-1.0.json",
    "title": "DataOps DataFlow Production Line",
    "type": "object",
    "additionalProperties": False,
    "required": [
        "schema_version",
        "dataflow_uid",
        "name",
        "input_schema_refs",
        "output_schema_ref",
        "components",
        "parameters",
    ],
}

RULE_CANDIDATE_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "https://dataops.local/schemas/rule-candidate-1.0.json",
    "title": "DataOps AI Rule Candidate",
    "type": "object",
    "additionalProperties": False,
    "required": sorted(CANDIDATE_KEYS - {"standard_spec"}),
    "properties": {
        "schema_version": {"const": SCHEMA_VERSION},
        "candidate_type": {"const": "rule"},
        "rule_spec": RULE_SPEC_SCHEMA,
        "standard_spec": {"type": "null"},
        "assumptions": {
            "type": "array",
            "maxItems": 50,
            "items": {"type": "string", "maxLength": 1000},
        },
        "ambiguities": {
            "type": "array",
            "maxItems": 50,
            "items": {"type": "string", "maxLength": 1000},
        },
        "confidence": {"type": "number", "minimum": 0, "maximum": 1},
        "explanation": {
            "type": "string",
            "minLength": 1,
            "maxLength": 2000,
        },
    },
}


def _closed_object(value: Any, allowed: set[str], label: str) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError(f"{label} must be an object")
    unknown = sorted(set(value) - allowed)
    if unknown:
        raise ValueError(
            f"{label} contains unsupported fields: {', '.join(unknown)}"
        )
    return value


def _required_string(value: Any, label: str, maximum: int = 500) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{label} is required")
    normalized = value.strip()
    if len(normalized) > maximum:
        raise ValueError(f"{label} exceeds {maximum} characters")
    return normalized


def _optional_string(value: Any, label: str, maximum: int = 2000) -> str:
    if not isinstance(value, str):
        raise ValueError(f"{label} must be a string")
    normalized = value.strip()
    if len(normalized) > maximum:
        raise ValueError(f"{label} exceeds {maximum} characters")
    return normalized


def _uid(value: Any, label: str) -> str:
    try:
        return ensure_governance_uid({"uid": str(value)})
    except ValueError as exc:
        raise ValueError(f"{label} must be a valid UUIDv7") from exc


def _identifier(value: Any, label: str) -> str:
    result = _required_string(value, label, 100)
    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,99}", result):
        raise ValueError(f"{label} contains unsupported characters")
    return result


def _normalized_secret_key(value: Any) -> str:
    return re.sub(r"[^a-z0-9]", "", str(value).lower())


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


def _bounded_strings(
    value: Any,
    label: str,
    *,
    maximum_items: int = 100,
    maximum_length: int = 1000,
) -> list[str]:
    if not isinstance(value, list) or len(value) > maximum_items:
        raise ValueError(f"{label} must be a bounded array")
    return [
        _required_string(item, f"{label} item", maximum_length) for item in value
    ]


def _canonical_hash(value: dict[str, Any]) -> str:
    canonical = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _validate_rule_step(value: Any) -> dict[str, Any]:
    step = copy.deepcopy(_closed_object(value, RULE_STEP_KEYS, "rule step"))
    step["id"] = _identifier(step.get("id"), "rule step id")
    operation = _required_string(step.get("op"), "rule step operation", 100)
    if operation not in RULE_OPS:
        raise ValueError(f"unsupported rule operation: {operation}")
    step["op"] = operation

    for key in ("column", "target", "to", "pattern", "replacement"):
        if key in step:
            step[key] = _required_string(
                step[key], f"rule step {key}", 2000
            )
    has_source_expression = "expression" in step
    has_canonical_expression = (
        "expression_text" in step or "expression_ast" in step
    )
    if has_source_expression and has_canonical_expression:
        raise ValueError("rule step expression cannot mix source and canonical forms")
    if has_source_expression:
        expression_text = _required_string(
            step.pop("expression"), "rule step expression", 20_000
        )
        step["expression_text"] = expression_text
        step["expression_ast"] = parse_expression(expression_text)
    elif has_canonical_expression:
        expression_text = _required_string(
            step.get("expression_text"), "rule step expression_text", 20_000
        )
        if "expression_ast" not in step:
            raise ValueError("rule step expression_ast is required")
        parsed_expression = parse_expression(expression_text)
        if step["expression_ast"] != parsed_expression:
            raise ValueError("rule step expression_ast must match expression_text")
        step["expression_text"] = expression_text
        step["expression_ast"] = parsed_expression
    for key in ("keys", "order_by", "group_by"):
        if key in step:
            step[key] = _bounded_strings(
                step[key], f"rule step {key}", maximum_length=200
            )
    for key in ("mapping", "aggregations", "lookup"):
        if key in step and not isinstance(step[key], dict):
            raise ValueError(f"rule step {key} must be an object")
    for key in ("trim", "lowercase", "uppercase"):
        if key in step and not isinstance(step[key], bool):
            raise ValueError(f"rule step {key} must be a boolean")
    if "severity" in step and step["severity"] not in SEVERITIES:
        raise ValueError("unsupported rule step severity")
    for key in ("on_error", "on_failure"):
        if key in step and step[key] not in FAILURE_ACTIONS:
            raise ValueError(f"unsupported rule step {key}")
    if "keep" in step and step["keep"] not in {"first", "last"}:
        raise ValueError("deduplicate keep must be first or last")
    if operation == "assert":
        if "expression_ast" not in step:
            raise ValueError("assert expression is required")
        if step.get("on_failure") not in FAILURE_ACTIONS:
            raise ValueError("assert on_failure is required")
    if operation == "cast":
        _required_string(step.get("column"), "cast column", 200)
        _required_string(step.get("to"), "cast target type", 100)
    if operation in {"normalize_text", "regex_replace", "fill_null"}:
        _required_string(step.get("column"), f"{operation} column", 200)
    if operation == "deduplicate" and not step.get("keys"):
        raise ValueError("deduplicate keys are required")
    if operation == "mask":
        _required_string(step.get("policy"), "mask policy", 200)
    _reject_secret_material(step)
    return step


def _normalize_rule_spec(value: Any, *, allow_legacy: bool) -> dict[str, Any]:
    spec = copy.deepcopy(_closed_object(value, RULE_ROOT_KEYS, "rule spec"))
    schema_version = spec.get("schema_version")
    if schema_version == LEGACY_RULE_SCHEMA_VERSION:
        if not allow_legacy:
            raise ValueError(
                f"rule spec schema_version must be {RULE_SCHEMA_VERSION}"
            )
    elif schema_version != RULE_SCHEMA_VERSION:
        raise ValueError(f"rule spec schema_version must be {RULE_SCHEMA_VERSION}")
    # Legacy assets are readable for migration, but canonical persistence and
    # every newly authored RuleSpec use V2 expression text plus JSON AST.
    spec["schema_version"] = RULE_SCHEMA_VERSION
    spec["rule_uid"] = _uid(spec.get("rule_uid"), "rule_uid")
    spec["name"] = _required_string(spec.get("name"), "rule name", 200)
    if "description" in spec:
        spec["description"] = _optional_string(
            spec["description"], "rule description"
        )
    spec["input_schema_ref"] = _required_string(
        spec.get("input_schema_ref"), "input_schema_ref"
    )
    spec["output_schema_ref"] = _required_string(
        spec.get("output_schema_ref"), "output_schema_ref"
    )
    raw_steps = spec.get("steps")
    if not isinstance(raw_steps, list) or not raw_steps or len(raw_steps) > 200:
        raise ValueError("rule steps must be a non-empty bounded array")
    spec["steps"] = [_validate_rule_step(item) for item in raw_steps]
    ids = [item["id"] for item in spec["steps"]]
    if len(ids) != len(set(ids)):
        raise ValueError("rule step ids must be unique")
    if spec.get("null_policy") not in NULL_POLICIES:
        raise ValueError("unsupported null_policy")
    spec["timezone"] = _required_string(
        spec.get("timezone"), "rule timezone", 100
    )
    _reject_secret_material(spec)
    return spec


def validate_rule_spec(value: Any) -> dict[str, Any]:
    """Validate a newly authored or newly persisted V2 RuleSpec."""

    return _normalize_rule_spec(value, allow_legacy=False)


def read_rule_spec(value: Any) -> dict[str, Any]:
    """Read a persisted RuleSpec, migrating legacy V1 only in memory."""

    return _normalize_rule_spec(value, allow_legacy=True)


def rule_spec_hash(value: Any) -> str:
    return _canonical_hash(read_rule_spec(value))


def validate_standard_spec(value: Any) -> dict[str, Any]:
    spec = copy.deepcopy(
        _closed_object(value, STANDARD_ROOT_KEYS, "standard spec")
    )
    if spec.get("schema_version") != SCHEMA_VERSION:
        raise ValueError(f"standard spec schema_version must be {SCHEMA_VERSION}")
    spec["standard_uid"] = _uid(spec.get("standard_uid"), "standard_uid")
    spec["name"] = _required_string(spec.get("name"), "standard name", 200)
    if "description" in spec:
        spec["description"] = _optional_string(
            spec["description"], "standard description"
        )
    scope = copy.deepcopy(
        _closed_object(spec.get("scope"), STANDARD_SCOPE_KEYS, "standard scope")
    )
    scope["object_type"] = _required_string(
        scope.get("object_type"), "standard scope object_type", 100
    )
    scope["schema_ref"] = _required_string(
        scope.get("schema_ref"), "standard scope schema_ref"
    )
    if "business_domain_uid" in scope:
        scope["business_domain_uid"] = _uid(
            scope["business_domain_uid"], "standard scope business_domain_uid"
        )
    spec["scope"] = scope
    raw_clauses = spec.get("clauses")
    if (
        not isinstance(raw_clauses, list)
        or not raw_clauses
        or len(raw_clauses) > 200
    ):
        raise ValueError("standard clauses must be a non-empty bounded array")
    clauses = []
    for raw_clause in raw_clauses:
        clause = copy.deepcopy(
            _closed_object(
                raw_clause, STANDARD_CLAUSE_KEYS, "standard clause"
            )
        )
        clause["id"] = _identifier(clause.get("id"), "standard clause id")
        clause["description"] = _required_string(
            clause.get("description"), "standard clause description", 2000
        )
        if clause.get("severity") not in SEVERITIES:
            raise ValueError("unsupported standard clause severity")
        clause["rule_version_id"] = _uid(
            clause.get("rule_version_id"), "standard clause rule_version_id"
        )
        if clause.get("exception_policy") not in FAILURE_ACTIONS:
            raise ValueError("unsupported standard clause exception_policy")
        clauses.append(clause)
    clause_ids = [item["id"] for item in clauses]
    if len(clause_ids) != len(set(clause_ids)):
        raise ValueError("standard clause ids must be unique")
    spec["clauses"] = clauses
    _reject_secret_material(spec)
    return spec


def standard_spec_hash(value: Any) -> str:
    return _canonical_hash(validate_standard_spec(value))


def _validate_idempotency(value: Any) -> dict[str, Any]:
    item = copy.deepcopy(
        _closed_object(value, IDEMPOTENCY_KEYS, "component idempotency")
    )
    if item.get("strategy") not in IDEMPOTENCY_STRATEGIES:
        raise ValueError("unsupported component idempotency strategy")
    item["key"] = _required_string(
        item.get("key"), "component idempotency key", 500
    )
    return item


def validate_dataflow_spec(value: Any) -> dict[str, Any]:
    spec = copy.deepcopy(
        _closed_object(value, DATAFLOW_ROOT_KEYS, "dataflow spec")
    )
    if spec.get("schema_version") != SCHEMA_VERSION:
        raise ValueError(f"dataflow spec schema_version must be {SCHEMA_VERSION}")
    spec["dataflow_uid"] = _uid(spec.get("dataflow_uid"), "dataflow_uid")
    spec["name"] = _required_string(spec.get("name"), "dataflow name", 200)
    if "description" in spec:
        spec["description"] = _optional_string(
            spec["description"], "dataflow description"
        )
    spec["input_schema_refs"] = _bounded_strings(
        spec.get("input_schema_refs"),
        "input_schema_refs",
        maximum_items=50,
        maximum_length=500,
    )
    if not spec["input_schema_refs"]:
        raise ValueError("input_schema_refs must not be empty")
    spec["output_schema_ref"] = _required_string(
        spec.get("output_schema_ref"), "output_schema_ref"
    )
    raw_components = spec.get("components")
    if (
        not isinstance(raw_components, list)
        or not raw_components
        or len(raw_components) > 500
    ):
        raise ValueError("dataflow components must be a non-empty bounded array")
    components = []
    for raw_component in raw_components:
        component = copy.deepcopy(
            _closed_object(
                raw_component, COMPONENT_KEYS, "dataflow component"
            )
        )
        component["id"] = _identifier(
            component.get("id"), "dataflow component id"
        )
        component_type = component.get("type")
        if component_type not in COMPONENT_TYPES:
            raise ValueError("unsupported dataflow component type")
        component["type"] = component_type
        if component.get("stage") not in STAGES:
            raise ValueError("unsupported dataflow component stage")
        order = component.get("order")
        if (
            isinstance(order, bool)
            or not isinstance(order, int)
            or order < 0
            or order > 1_000_000
        ):
            raise ValueError("dataflow component order must be a bounded integer")
        if component_type == "standard.enforce":
            component["standard_version_id"] = _uid(
                component.get("standard_version_id"),
                "component standard_version_id",
            )
            if "rule_version_id" in component or "idempotency" in component:
                raise ValueError(
                    "standard.enforce cannot define rule or idempotency"
                )
        else:
            component["rule_version_id"] = _uid(
                component.get("rule_version_id"),
                "component rule_version_id",
            )
            if "standard_version_id" in component:
                raise ValueError(
                    "rule component cannot define standard_version_id"
                )
            if component_type == "rule.apply":
                component["idempotency"] = _validate_idempotency(
                    component.get("idempotency")
                )
            elif "idempotency" in component:
                raise ValueError("quality.check cannot define idempotency")
        components.append(component)
    component_ids = [item["id"] for item in components]
    if len(component_ids) != len(set(component_ids)):
        raise ValueError("dataflow component ids must be unique")
    spec["components"] = sorted(
        components, key=lambda item: (item["order"], item["id"])
    )
    parameters = spec.get("parameters")
    if not isinstance(parameters, dict) or len(parameters) > 100:
        raise ValueError("dataflow parameters must be a bounded object")
    spec["parameters"] = parameters
    _reject_secret_material(spec)
    return spec


def dataflow_spec_hash(value: Any) -> str:
    return _canonical_hash(validate_dataflow_spec(value))


def validate_rule_candidate(value: Any) -> dict[str, Any]:
    candidate = copy.deepcopy(
        _closed_object(value, CANDIDATE_KEYS, "rule candidate")
    )
    missing = sorted(
        (CANDIDATE_KEYS - {"rule_spec", "standard_spec"}) - set(candidate)
    )
    if missing:
        raise ValueError(
            f"rule candidate is missing fields: {', '.join(missing)}"
        )
    if candidate.get("schema_version") != SCHEMA_VERSION:
        raise ValueError(
            f"rule candidate schema_version must be {SCHEMA_VERSION}"
        )
    candidate_type = candidate.get("candidate_type")
    if candidate_type not in {"rule", "standard"}:
        raise ValueError("unsupported rule candidate_type")
    if candidate_type == "rule":
        if "rule_spec" not in candidate:
            raise ValueError("rule candidate is missing fields: rule_spec")
        candidate["rule_spec"] = validate_rule_spec(
            candidate.get("rule_spec")
        )
        candidate.setdefault("standard_spec", None)
        if candidate.get("standard_spec") is not None:
            raise ValueError("rule candidate cannot contain standard_spec")
    else:
        if "standard_spec" not in candidate:
            raise ValueError("standard candidate is missing fields: standard_spec")
        candidate["standard_spec"] = validate_standard_spec(
            candidate.get("standard_spec")
        )
        if candidate.get("rule_spec") is not None:
            candidate["rule_spec"] = validate_rule_spec(
                candidate["rule_spec"]
            )
        else:
            candidate["rule_spec"] = None
    candidate["assumptions"] = _bounded_strings(
        candidate.get("assumptions"), "candidate assumptions", maximum_items=50
    )
    candidate["ambiguities"] = _bounded_strings(
        candidate.get("ambiguities"), "candidate ambiguities", maximum_items=50
    )
    confidence = candidate.get("confidence")
    if (
        isinstance(confidence, bool)
        or not isinstance(confidence, (int, float))
        or confidence < 0
        or confidence > 1
    ):
        raise ValueError("candidate confidence must be between 0 and 1")
    candidate["confidence"] = float(confidence)
    candidate["explanation"] = _required_string(
        candidate.get("explanation"), "candidate explanation", 2000
    )
    _reject_secret_material(candidate)
    return candidate