"""Governed control-plane endpoints for AI-authored data rules."""

from __future__ import annotations

import hashlib
import json
from datetime import UTC, datetime, timedelta
from typing import Any

from flask import current_app, g, jsonify, request

from app import db
from app.api.data_rules import bp
from app.core.data_rules.authoring import (
    OpenAICompatibleRuleModel,
    RuleAuthoringAgent,
)
from app.core.data_rules.contracts import (
    dataflow_spec_hash,
    rule_spec_hash,
    standard_spec_hash,
    validate_dataflow_spec,
    validate_rule_spec,
    validate_standard_spec,
)
from app.core.data_rules.deployment import (
    DataFlowDeploymentService,
    DeploymentOperationError,
)
from app.core.data_rules.production_line import resolve_production_line
from app.core.data_rules.publication import (
    GenerationReceiptSigner,
    LogicalRuleCompiler,
    PhysicalPlanPublicationService,
    RulePublicationService,
    RuleValidationRejected,
    ServerOwnedLogicalDryRunRunner,
    ServerOwnedPhysicalPreflightRunner,
    generation_receipt_claims,
)
from app.core.data_rules.release import ProductionLineReleaseService
from app.core.data_rules.repository import DataRuleRepository
from app.core.data_rules.schema_resolver import (
    Neo4jSchemaMetadataCatalog,
    SchemaResolver,
)
from app.core.orchestration.engines.kestra import KestraAdapter
from app.models.result import failed, success
from app.runner.auth import TaskTokenIssuer

_VALIDATORS = {
    "rule": (validate_rule_spec, rule_spec_hash),
    "standard": (validate_standard_spec, standard_spec_hash),
    "dataflow": (validate_dataflow_spec, dataflow_spec_hash),
}


def _body() -> dict[str, Any]:
    value = request.get_json(silent=True)
    if not isinstance(value, dict):
        raise ValueError("request body must be an object")
    return value


def _bad_request(message: str = "规则请求无效"):
    return jsonify(failed(message, code=400)), 400


def _closed_body(allowed: set[str]) -> dict[str, Any]:
    body = _body()
    if set(body) - allowed:
        raise ValueError("request contains unsupported fields")
    return body


def _repository() -> DataRuleRepository:
    configured = current_app.extensions.get("data_rule_repository")
    if configured is not None:
        return configured
    return DataRuleRepository(db.session)


def _release_service() -> ProductionLineReleaseService:
    configured = current_app.extensions.get("production_line_release_service")
    if configured is not None:
        return configured
    repository = _repository()
    resolver = current_app.extensions.get("data_rule_schema_resolver")
    if resolver is None:
        catalog = current_app.extensions.get("data_rule_metadata_catalog")
        resolver = SchemaResolver(catalog or Neo4jSchemaMetadataCatalog(), repository)
    return ProductionLineReleaseService(repository, schema_resolver=resolver)


def _schema_resolver() -> SchemaResolver:
    configured = current_app.extensions.get("data_rule_schema_resolver")
    if configured is not None:
        return configured
    repository = _repository()
    catalog = current_app.extensions.get("data_rule_metadata_catalog")
    return SchemaResolver(catalog or Neo4jSchemaMetadataCatalog(), repository)


def _deployment_readiness() -> dict[str, Any]:
    reasons = []
    if not current_app.config.get("KESTRA_BASE_URL"):
        reasons.append("kestra_base_url_missing")
    if not current_app.config.get("KESTRA_USERNAME"):
        reasons.append("kestra_username_missing")
    if not current_app.config.get("KESTRA_PASSWORD"):
        reasons.append("kestra_password_missing")
    secret = current_app.config.get("RUNNER_TASK_TOKEN_SECRET")
    if not isinstance(secret, str) or len(secret.encode("utf-8")) < 32:
        reasons.append("runner_task_token_secret_invalid")
    gate_open = bool(current_app.config.get("DATA_FACTORY_ACTIVATION_ENABLED", False))
    if not gate_open:
        reasons.append("post_acceptance_activation_gate_closed")
    return {
        "ready": not reasons,
        "gate_open": gate_open,
        "reasons": reasons,
        "server_owned": True,
    }


def _deployment_service() -> DataFlowDeploymentService:
    configured = current_app.extensions.get("dataflow_deployment_service")
    if configured is not None:
        return configured
    readiness = _deployment_readiness()
    infrastructure_reasons = [
        reason
        for reason in readiness["reasons"]
        if reason != "post_acceptance_activation_gate_closed"
    ]
    if infrastructure_reasons:
        raise RuntimeError("data factory deployment infrastructure is not ready")
    engine = KestraAdapter(
        current_app.config["KESTRA_BASE_URL"],
        current_app.config["KESTRA_USERNAME"],
        current_app.config["KESTRA_PASSWORD"],
        tenant_id=current_app.config.get("KESTRA_TENANT_ID", "main"),
        timeout=int(current_app.config.get("KESTRA_HTTP_TIMEOUT_SECONDS", 30)),
    )
    service = DataFlowDeploymentService(
        _repository(),
        engine=engine,
        token_issuer=TaskTokenIssuer(current_app.config["RUNNER_TASK_TOKEN_SECRET"]),
    )
    current_app.extensions["dataflow_deployment_service"] = service
    return service


def _activation_gate_ready() -> bool:
    if current_app.extensions.get("dataflow_deployment_service") is not None:
        return bool(current_app.config.get("DATA_FACTORY_ACTIVATION_ENABLED", False))
    return bool(_deployment_readiness()["ready"])


def _receipt_signer() -> GenerationReceiptSigner:
    configured = current_app.extensions.get("generation_receipt_signer")
    if configured is not None:
        return configured
    secret = current_app.config.get("RULE_GENERATION_RECEIPT_SECRET")
    if not isinstance(secret, str) or not secret.strip():
        raise RuntimeError("dedicated generation receipt secret is not configured")
    signer = GenerationReceiptSigner(secret)
    current_app.extensions["generation_receipt_signer"] = signer
    return signer


def _rule_artifact_store():
    configured = current_app.extensions.get("rule_artifact_store")
    if configured is not None:
        return configured
    from minio import Minio

    from app.runner.artifacts import ArtifactStore

    store = ArtifactStore(
        Minio(
            current_app.config["MINIO_HOST"],
            access_key=current_app.config["MINIO_USER"],
            secret_key=current_app.config["MINIO_PASSWORD"],
            secure=bool(current_app.config["MINIO_SECURE"]),
        ),
        bucket=current_app.config["MINIO_BUCKET"],
        max_artifact_bytes=32 * 1024 * 1024,
        max_rows=100_000,
        memory_limit_bytes=256 * 1024 * 1024,
        max_ttl_seconds=3600,
    )
    current_app.extensions["rule_artifact_store"] = store
    return store


def _publication_service() -> RulePublicationService:
    configured = current_app.extensions.get("rule_publication_service")
    if configured is not None:
        return configured
    test_runner = current_app.extensions.get("rule_validation_test_runner")
    if test_runner is None:
        try:
            test_runner = ServerOwnedLogicalDryRunRunner(_rule_artifact_store())
        except Exception as exc:
            raise RuntimeError(
                "trusted rule validation test runner is not configured"
            ) from exc
        current_app.extensions["rule_validation_test_runner"] = test_runner
    service = RulePublicationService(
        _repository(),
        receipt_signer=_receipt_signer(),
        compiler=LogicalRuleCompiler(),
        test_runner=test_runner,
    )
    current_app.extensions["rule_publication_service"] = service
    return service


def _physical_publication_service() -> PhysicalPlanPublicationService:
    configured = current_app.extensions.get("physical_plan_publication_service")
    if configured is not None:
        return configured
    test_runner = current_app.extensions.get("rule_physical_test_runner")
    if test_runner is None:
        try:
            from app.core.data_source.runtime import get_data_source_manager

            test_runner = ServerOwnedPhysicalPreflightRunner(
                _rule_artifact_store(),
                datasource_manager=get_data_source_manager(),
            )
        except Exception as exc:
            raise RuntimeError(
                "trusted physical plan test runner is not configured"
            ) from exc
        current_app.extensions["rule_physical_test_runner"] = test_runner
    service = PhysicalPlanPublicationService(_repository(), test_runner=test_runner)
    current_app.extensions["physical_plan_publication_service"] = service
    return service


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


@bp.get("/capabilities")
def capabilities():
    readiness = _deployment_readiness()
    if current_app.extensions.get("dataflow_deployment_service") is not None:
        readiness = {
            **readiness,
            "ready": bool(readiness["gate_open"]),
            "reasons": [
                reason
                for reason in readiness["reasons"]
                if reason == "post_acceptance_activation_gate_closed"
            ],
        }
    return jsonify(
        success(
            {
                "natural_language_authoring": True,
                "schema_constrained_candidates": True,
                "production_line_preview": True,
                "immutable_asset_versions": True,
                "server_side_publishing": True,
                "production_line_release": True,
                "data_factory_activation": readiness["ready"],
                "data_factory_readiness": readiness,
            }
        )
    )


@bp.post("/production-lines/draft-identity")
def create_production_line_draft_identity():
    """Persist a short-lived, single-use, actor-bound DataFlow reservation."""
    try:
        _closed_body(set())
        result = _repository().reserve_dataflow_draft(actor_uid=g.current_user["id"])
        db.session.commit()
        return (
            jsonify(
                success(
                    result,
                    "生产线草稿身份已创建",
                    code=201,
                )
            ),
            201,
        )
    except (TypeError, ValueError):
        db.session.rollback()
        return _bad_request("生产线草稿身份请求无效")
    except Exception:
        db.session.rollback()
        current_app.logger.exception("create DataFlow draft reservation failed")
        return jsonify(failed("生产线草稿身份暂时不可用", code=503)), 503


@bp.post("/validate")
def validate_asset():
    try:
        body = _closed_body({"asset_type", "spec"})
        asset_type = body.get("asset_type")
        if asset_type not in _VALIDATORS:
            raise ValueError("unsupported asset type")
        validator, hasher = _VALIDATORS[asset_type]
        normalized = validator(body.get("spec"))
        return jsonify(
            success(
                {
                    "asset_type": asset_type,
                    "normalized": normalized,
                    "spec_hash": hasher(normalized),
                }
            )
        )
    except (TypeError, ValueError):
        return _bad_request("规则定义无效")


def _authoring_agent() -> RuleAuthoringAgent:
    configured = current_app.extensions.get("data_rule_authoring_agent")
    if configured is not None:
        return configured
    api_key = current_app.config.get("LLM_API_KEY") or current_app.config.get(
        "DEEPSEEK_API_KEY"
    )
    if not api_key:
        raise RuntimeError("rule authoring model is not configured")
    agent = RuleAuthoringAgent(model=OpenAICompatibleRuleModel())
    current_app.extensions["data_rule_authoring_agent"] = agent
    return agent


@bp.post("/interpret")
def interpret_rule():
    try:
        body = _closed_body({"source_text", "authoring_surface", "context"})
        receipt_signer = _receipt_signer()
        repository = _repository()
        validation_context = repository.resolve_validation_context(
            body.get("context", {})
        )
        result = _authoring_agent().interpret(
            source_text=body.get("source_text"),
            authoring_surface=body.get("authoring_surface"),
            context=validation_context,
        )
        candidate = result.get("candidate")
        if body.get("authoring_surface") == "data_standard" and (
            not isinstance(candidate, dict)
            or candidate.get("candidate_type") != "rule"
            or not isinstance(candidate.get("rule_spec"), dict)
        ):
            raise ValueError("data_standard authoring must produce a rule candidate")
        audit = repository.record_generation_run(
            evidence=result,
            created_by=g.current_user["id"],
            validation_context=validation_context,
        )
        result = {
            **result,
            "generation_run_id": audit["id"],
            "correlation_id": audit["correlation_id"],
        }
        if (
            result.get("status") == "ready"
            and isinstance(candidate, dict)
            and candidate.get("candidate_type") == "rule"
            and isinstance(candidate.get("rule_spec"), dict)
        ):
            claims = generation_receipt_claims(
                generation_run_id=audit["id"],
                actor_uid=g.current_user["id"],
                source_text=result["source_text"],
                candidate_hash=result["candidate_hash"],
                rule_spec=candidate["rule_spec"],
                model_hash=result.get("model_hash")
                or _metadata_hash(
                    {
                        "provider": result.get("model_provider", "unknown"),
                        "name": result.get("model_name", "unknown"),
                    }
                ),
                prompt_hash=result.get("prompt_hash")
                or _metadata_hash(result.get("prompt_version", "unknown")),
                context_hash=result["context_hash"],
                expires_at=datetime.now(UTC) + timedelta(minutes=10),
            )
            result["generation_receipt"] = receipt_signer.issue(claims)
        db.session.commit()
        return jsonify(success(result))
    except (TypeError, ValueError):
        db.session.rollback()
        return _bad_request("自然语言规则描述无效")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("AI 规则解析服务未配置", code=503)), 503
    except Exception:
        db.session.rollback()
        current_app.logger.exception("AI rule interpretation failed")
        return jsonify(failed("AI 规则解析暂时不可用", code=503)), 503


@bp.post("/rule-versions")
def create_rule_version():
    try:
        body = _closed_body(
            {
                "rule_spec",
                "source_text",
                "generation_receipt",
                "category",
                "source_language",
                "generated_kind",
            }
        )
        # Enforce V2 at the HTTP boundary even when a test/different
        # repository implementation is injected.
        rule_spec = validate_rule_spec(body.get("rule_spec"))
        result = _publication_service().create_draft(
            rule_spec=rule_spec,
            source_text=body.get("source_text"),
            category=body.get("category", "general"),
            source_language=body.get("source_language", "zh-CN"),
            generated_kind=body.get("generated_kind", "rulespec"),
            actor_uid=g.current_user["id"],
            generation_receipt=body.get("generation_receipt"),
        )
        db.session.commit()
        return jsonify(success(result, "规则版本创建成功")), 201
    except (TypeError, ValueError):
        db.session.rollback()
        return _bad_request("规则版本定义无效")
    except Exception:
        db.session.rollback()
        current_app.logger.exception("create rule version failed")
        return jsonify(failed("规则版本创建失败", code=500)), 500


@bp.post("/rule-versions/<version_id>/publish")
def publish_rule_version(version_id: str):
    try:
        if request.get_data(cache=True) and request.get_json(silent=True) not in (
            None,
            {},
        ):
            raise ValueError("publish request must not contain fields")
        result = _publication_service().publish(version_id, g.current_user["id"])
        db.session.commit()
        return jsonify(success(result, "规则版本发布成功"))
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("规则版本无法发布", code=409)), 409
    except Exception:
        db.session.rollback()
        current_app.logger.exception("publish rule version failed")
        return jsonify(failed("规则版本发布失败", code=500)), 500


@bp.post("/rule-versions/<version_id>/validate")
def validate_rule_version(version_id: str):
    try:
        if request.get_data(cache=True) and request.get_json(silent=True) not in (
            None,
            {},
        ):
            raise ValueError("validate request must not contain fields")
        result = _publication_service().validate(version_id, g.current_user["id"])
        db.session.commit()
        return jsonify(success(result, "规则版本编译验证成功"))
    except RuleValidationRejected:
        db.session.commit()
        return jsonify(failed("规则版本编译验证失败", code=409)), 409
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("规则版本编译验证失败", code=409)), 409
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("规则验证服务未配置", code=503)), 503


@bp.post("/rule-versions/<version_id>/test")
def test_rule_version(version_id: str):
    try:
        body = _closed_body({"plan_id"})
        result = _publication_service().test(
            version_id,
            g.current_user["id"],
            plan_id=body.get("plan_id"),
        )
        db.session.commit()
        return jsonify(success(result, "规则版本样本测试成功"))
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("规则版本样本测试失败", code=409)), 409
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("规则测试服务未配置", code=503)), 503


@bp.get("/rule-versions/<version_id>/evidence")
def rule_version_evidence(version_id: str):
    try:
        return jsonify(
            success(
                _repository().get_asset_evidence(
                    asset_type="rule", version_id=version_id
                )
            )
        )
    except (TypeError, ValueError):
        return jsonify(failed("规则证据不存在", code=404)), 404
    except Exception:
        current_app.logger.exception("load rule evidence failed")
        return jsonify(failed("规则证据暂时不可用", code=503)), 503


@bp.get("/catalog/assets/<asset_type>/<version_id>/evidence")
def catalog_asset_evidence(asset_type: str, version_id: str):
    try:
        return jsonify(
            success(
                _repository().get_asset_evidence(
                    asset_type=asset_type,
                    version_id=version_id,
                )
            )
        )
    except (TypeError, ValueError):
        return jsonify(failed("资产证据不存在", code=404)), 404
    except Exception:
        current_app.logger.exception("load catalog asset evidence failed")
        return jsonify(failed("资产证据暂时不可用", code=503)), 503


def _catalog_schema_context():
    raw_inputs = request.args.get("input_schema_refs")
    output = request.args.get("output_schema_ref")
    if raw_inputs is None and output is None:
        return None, None
    if raw_inputs is None or output is None:
        raise ValueError("catalog schema context is incomplete")
    inputs = json.loads(raw_inputs)
    if not isinstance(inputs, list):
        raise ValueError("catalog input_schema_refs must be an array")
    return inputs, output


@bp.get("/catalog/assets/<asset_type>/<version_id>")
def catalog_asset(asset_type: str, version_id: str):
    try:
        if set(request.args) - {
            "input_schema_refs",
            "output_schema_ref",
        }:
            raise ValueError("catalog asset query contains unsupported fields")
        inputs, output = _catalog_schema_context()
        result = _repository().get_published_asset(
            asset_type=asset_type,
            version_id=version_id,
            input_schema_refs=inputs,
            output_schema_ref=output,
            schema_resolver=_schema_resolver() if inputs is not None else None,
        )
        if inputs is not None:
            # SchemaResolver is a read-through snapshot boundary. Persist the
            # IDs returned to this response before making them observable.
            db.session.commit()
        return jsonify(success(result))
    except (TypeError, ValueError, json.JSONDecodeError):
        db.session.rollback()
        return jsonify(failed("已发布资产不存在或上下文无效", code=404)), 404
    except Exception:
        db.session.rollback()
        current_app.logger.exception("load exact catalog asset failed")
        return jsonify(failed("规则目录暂时不可用", code=503)), 503


@bp.get("/catalog")
@bp.get("/catalog/rule-versions")
def published_rule_catalog():
    try:
        legacy_rule_alias = request.path.endswith("/rule-versions")
        allowed = {
            "query",
            "limit",
            "offset",
            "input_schema_refs",
            "output_schema_ref",
        }
        if not legacy_rule_alias:
            allowed.add("asset_type")
        if set(request.args) - allowed:
            raise ValueError("catalog query contains unsupported fields")
        query = request.args.get("query", "")
        limit = int(request.args.get("limit", "50"))
        offset = int(request.args.get("offset", "0"))
        asset_type = (
            "rule" if legacy_rule_alias else request.args.get("asset_type") or None
        )
        if asset_type not in {None, "rule", "standard"}:
            raise ValueError("catalog asset_type is invalid")
        if len(query) > 200 or limit < 1 or limit > 100:
            raise ValueError("catalog bounds are invalid")
        if offset < 0 or offset > 1_000_000:
            raise ValueError("catalog offset is invalid")
        inputs, output = _catalog_schema_context()
        catalog_args = {
            "query": query,
            "asset_type": asset_type,
            "limit": limit,
            "offset": offset,
        }
        if inputs is not None:
            catalog_args.update(
                {
                    "input_schema_refs": inputs,
                    "output_schema_ref": output,
                    "schema_resolver": _schema_resolver(),
                }
            )
        result = _repository().search_published_assets(**catalog_args)
        if inputs is not None:
            db.session.commit()
        return jsonify(success(result))
    except (TypeError, ValueError):
        db.session.rollback()
        return _bad_request("规则目录查询无效")
    except Exception:
        db.session.rollback()
        current_app.logger.exception("load rule catalog failed")
        return jsonify(failed("规则目录暂时不可用", code=503)), 503


@bp.post("/execution-plans/<plan_id>/validate")
def validate_physical_plan(plan_id: str):
    try:
        if request.get_data(cache=True) and request.get_json(silent=True) not in (
            None,
            {},
        ):
            raise ValueError("validate request must not contain fields")
        result = _physical_publication_service().validate(plan_id, g.current_user["id"])
        db.session.commit()
        return jsonify(success(result, "物理执行计划编译证据已确认"))
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("物理执行计划验证失败", code=409)), 409
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("物理计划验证服务未配置", code=503)), 503


@bp.post("/execution-plans/<plan_id>/test")
def test_physical_plan(plan_id: str):
    try:
        if request.get_data(cache=True) and request.get_json(silent=True) not in (
            None,
            {},
        ):
            raise ValueError("test request must not contain fields")
        result = _physical_publication_service().test(plan_id, g.current_user["id"])
        db.session.commit()
        return jsonify(success(result, "物理执行计划样本测试成功"))
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("物理执行计划样本测试失败", code=409)), 409
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("物理计划测试服务未配置", code=503)), 503


@bp.post("/execution-plans/<plan_id>/publish")
def publish_physical_plan(plan_id: str):
    try:
        if request.get_data(cache=True) and request.get_json(silent=True) not in (
            None,
            {},
        ):
            raise ValueError("publish request must not contain fields")
        result = _physical_publication_service().publish(plan_id, g.current_user["id"])
        db.session.commit()
        return jsonify(success(result, "物理执行计划发布成功"))
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("物理执行计划无法发布", code=409)), 409
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("物理计划发布服务未配置", code=503)), 503


@bp.post("/standard-versions")
def create_standard_version():
    try:
        body = _closed_body({"standard_spec", "source_text"})
        result = _repository().create_standard_version(
            standard_spec=body.get("standard_spec"),
            source_text=body.get("source_text"),
            created_by=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "数据标准版本创建成功")), 201
    except (TypeError, ValueError):
        db.session.rollback()
        return _bad_request("数据标准版本定义无效")
    except Exception:
        db.session.rollback()
        current_app.logger.exception("create standard version failed")
        return jsonify(failed("数据标准版本创建失败", code=500)), 500


@bp.post("/standard-versions/<version_id>/publish")
def publish_standard_version(version_id: str):
    try:
        if request.get_data(cache=True) and request.get_json(silent=True) not in (
            None,
            {},
        ):
            raise ValueError("publish request must not contain fields")
        result = _repository().publish_standard_version(
            version_id=version_id,
            published_by=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "数据标准版本发布成功"))
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("数据标准版本无法发布", code=409)), 409
    except Exception:
        db.session.rollback()
        current_app.logger.exception("publish standard version failed")
        return jsonify(failed("数据标准版本发布失败", code=500)), 500


@bp.post("/production-lines/resolve")
def resolve_production_line_preview():
    try:
        body = _body()
        package = resolve_production_line(
            body.get("dataflow_spec"),
            body.get("standard_versions"),
            body.get("rule_versions"),
            component_binding_ids=body.get("component_binding_ids"),
        )
        return jsonify(
            success(
                {
                    "preview": True,
                    "release_ready": False,
                    "package": package,
                }
            )
        )
    except (TypeError, ValueError):
        return _bad_request("数据生产线定义无效")


@bp.post("/production-lines/<dataflow_uid>/release")
def release_production_line(dataflow_uid: str):
    try:
        body = _closed_body(
            {
                "dataflow_spec",
                "source_text",
            }
        )
        result = _release_service().release(
            dataflow_uid=dataflow_uid,
            dataflow_spec=body.get("dataflow_spec"),
            source_text=body.get("source_text"),
            created_by=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "数据生产线发布成功")), 201
    except (TypeError, ValueError):
        db.session.rollback()
        return jsonify(failed("数据生产线无法发布", code=409)), 409
    except Exception:
        db.session.rollback()
        current_app.logger.exception("release production line failed")
        return jsonify(failed("数据生产线发布失败", code=500)), 500


@bp.get("/deployments")
def list_dataflow_deployments():
    try:
        if set(request.args) - {"environment"}:
            raise ValueError("deployment query contains unsupported fields")
        return jsonify(
            success(
                {
                    "items": _deployment_service().list(
                        environment=request.args.get("environment") or None
                    ),
                    "readiness": _deployment_readiness(),
                }
            )
        )
    except (TypeError, ValueError):
        return _bad_request("数据工厂投产查询无效")
    except RuntimeError:
        return jsonify(failed("数据工厂投产服务未配置", code=503)), 503


@bp.get("/deployments/candidates")
def list_dataflow_deployment_candidates():
    try:
        if set(request.args) != {"environment"}:
            raise ValueError("candidate environment is required")
        return jsonify(
            success(
                {
                    "items": _deployment_service().candidates(
                        environment=request.args["environment"]
                    )
                }
            )
        )
    except (TypeError, ValueError):
        return _bad_request("可入厂生产线查询无效")
    except RuntimeError:
        return jsonify(failed("数据工厂投产服务未配置", code=503)), 503


@bp.post("/deployments")
def create_dataflow_deployment():
    try:
        body = _closed_body(
            {
                "dataflow_version_id",
                "environment",
                "binding_snapshot",
                "schedule_plan",
                "reason",
                "idempotency_key",
                "correlation_id",
            }
        )
        result = _deployment_service().create(
            body.get("dataflow_version_id"),
            binding_snapshot=body.get("binding_snapshot"),
            environment=body.get("environment"),
            schedule_plan=body.get("schedule_plan"),
            actor_uid=g.current_user["id"],
            reason=body.get("reason"),
            idempotency_key=body.get("idempotency_key"),
            correlation_id=body.get("correlation_id"),
        )
        db.session.commit()
        return jsonify(success(result, "数据生产线已进入数据工厂")), 201
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "数据工厂投产定义无效")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("数据工厂投产服务未配置", code=503)), 503
    except Exception:
        db.session.rollback()
        current_app.logger.exception("create dataflow deployment failed")
        return jsonify(failed("数据工厂投产创建失败", code=500)), 500


def _deployment_action_body(extra: set[str] | None = None):
    return _closed_body(
        {
            "reason",
            "idempotency_key",
            "correlation_id",
            *(extra or set()),
        }
    )


def _deployment_conflict(exc: Exception, message: str):
    code = getattr(exc, "code", "terminal_conflict")
    disposition = getattr(exc, "idempotency_key_disposition", "rotate")
    error = {
        "code": code,
        "idempotency_key_disposition": disposition,
        "reconcile_required": code == "operation_unknown",
    }
    blocking_operation = getattr(exc, "blocking_operation", None)
    if blocking_operation is not None:
        error["blocking_operation"] = blocking_operation
    return (
        jsonify(
            failed(
                message,
                code=409,
                error=error,
            )
        ),
        409,
    )


@bp.post("/deployments/<deployment_id>/deploy-disabled")
def deploy_dataflow_disabled(deployment_id: str):
    try:
        body = _deployment_action_body()
        result = _deployment_service().deploy_disabled(
            deployment_id,
            g.current_user["id"],
            reason=body.get("reason", "deploy disabled"),
            idempotency_key=body.get("idempotency_key"),
            correlation_id=body.get("correlation_id"),
        )
        db.session.commit()
        return jsonify(success(result, "生产线已禁用部署"))
    except DeploymentOperationError as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产线禁用部署状态冲突")
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产线无法禁用部署")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("调度引擎部署结果异常", code=503)), 503


@bp.post("/deployments/<deployment_id>/canary")
def run_dataflow_canary(deployment_id: str):
    try:
        body = _deployment_action_body({"inputs"})
        result = _deployment_service().run_canary(
            deployment_id,
            body.get("inputs", {}),
            g.current_user["id"],
            reason=body.get("reason", "trial production"),
            idempotency_key=body.get("idempotency_key"),
            correlation_id=body.get("correlation_id"),
        )
        db.session.commit()
        return jsonify(success(result, "试生产已完成"))
    except DeploymentOperationError as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "试生产操作状态冲突")
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "试生产请求无效")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("试生产执行结果异常", code=503)), 503


@bp.post("/deployments/<deployment_id>/activate")
def activate_dataflow_deployment(deployment_id: str):
    if not _activation_gate_ready():
        return jsonify(failed("数据工厂激活门禁尚未开放", code=503)), 503
    try:
        body = _deployment_action_body({"evidence_id"})
        result = _deployment_service().activate(
            deployment_id,
            body.get("evidence_id"),
            g.current_user["id"],
            reason=body.get("reason", "mass production activation"),
            idempotency_key=body.get("idempotency_key"),
            correlation_id=body.get("correlation_id"),
        )
        db.session.commit()
        return jsonify(success(result, "生产线已转入批量生产"))
    except DeploymentOperationError as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产线激活状态冲突")
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产线无法激活")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("生产线激活结果异常", code=503)), 503


@bp.post("/deployments/<deployment_id>/execute")
def execute_active_dataflow_deployment(deployment_id: str):
    try:
        body = _deployment_action_body({"inputs"})
        result = _deployment_service().execute_active(
            deployment_id,
            body.get("inputs", {}),
            g.current_user["id"],
            reason=body.get("reason", "manual production execution"),
            idempotency_key=body.get("idempotency_key"),
            correlation_id=body.get("correlation_id"),
        )
        db.session.commit()
        return jsonify(success(result, "生产数据处理已完成"))
    except DeploymentOperationError as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产执行操作状态冲突")
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产执行请求无效")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("生产数据处理执行异常", code=503)), 503


@bp.post("/deployments/<deployment_id>/rollback")
def rollback_dataflow_deployment(deployment_id: str):
    try:
        body = _deployment_action_body()
        result = _deployment_service().rollback(
            deployment_id,
            g.current_user["id"],
            reason=body.get("reason", "restore previous production line"),
            idempotency_key=body.get("idempotency_key"),
            correlation_id=body.get("correlation_id"),
        )
        db.session.commit()
        return jsonify(success(result, "已恢复上一条稳定生产线"))
    except DeploymentOperationError as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产线回滚状态冲突")
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "生产线无法回滚")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("生产线回滚结果异常", code=503)), 503


def _reconcile_dataflow_deployment(deployment_id: str, action: str):
    try:
        body = _closed_body({"idempotency_key"})
        result = _deployment_service().reconcile(
            deployment_id,
            action,
            g.current_user["id"],
            idempotency_key=body.get("idempotency_key"),
        )
        db.session.commit()
        return jsonify(success(result, "投产未知结果已完成对账"))
    except PermissionError:
        db.session.rollback()
        return jsonify(failed("权限不足", code=403)), 403
    except DeploymentOperationError as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "投产对账状态冲突")
    except (TypeError, ValueError) as exc:
        db.session.rollback()
        return _deployment_conflict(exc, "投产未知结果无法对账")
    except RuntimeError:
        db.session.rollback()
        return jsonify(failed("投产对账服务异常", code=503)), 503


@bp.post("/deployments/<deployment_id>/reconcile-deploy")
def reconcile_dataflow_deploy(deployment_id: str):
    return _reconcile_dataflow_deployment(deployment_id, "deploy_disabled")


@bp.post("/deployments/<deployment_id>/reconcile-activate")
def reconcile_dataflow_activate(deployment_id: str):
    return _reconcile_dataflow_deployment(deployment_id, "activate")


@bp.post("/deployments/<deployment_id>/reconcile-rollback")
def reconcile_dataflow_rollback(deployment_id: str):
    return _reconcile_dataflow_deployment(deployment_id, "rollback")


@bp.post("/deployments/<deployment_id>/reconcile-execute")
def reconcile_dataflow_execute(deployment_id: str):
    return _reconcile_dataflow_deployment(deployment_id, "execute_active")