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

"""Governed Data Factory deployment lifecycle.

This service is the boundary between immutable production-line semantics and
environment-specific infrastructure.  It never accepts executable source,
recompiles a rule, or mutates a released package.
"""

from __future__ import annotations

import copy
import hashlib
import json
import re
import time
from collections.abc import Callable, Mapping
from datetime import UTC, datetime, timedelta
from typing import Any

import requests

from app.core.common.identifiers import ensure_governance_uid, new_governance_uid
from app.core.orchestration.compilers.kestra import compile_kestra_flow
from app.core.orchestration.engines.kestra import (
    canonical_flow_definition_hash,
)
from app.core.orchestration.spec import validate_schedule_plan

ENVIRONMENTS = frozenset({"development", "test", "production"})
KESTRA_HASH_VERSION = "kestra_canonical_v2"
TRANSITIONS = {
    "draft": frozenset({"disabled", "failed"}),
    "disabled": frozenset({"canary", "failed"}),
    "canary": frozenset({"active", "failed", "rolled_back"}),
    "active": frozenset({"superseded", "rolled_back", "failed"}),
    "failed": frozenset({"disabled", "rolled_back"}),
    "superseded": frozenset({"rolled_back", "active"}),
    "rolled_back": frozenset(),
}
_FORBIDDEN_INPUT_KEYS = frozenset(
    {
        "sql",
        "python",
        "script",
        "code",
        "yaml",
        "workflow_spec",
        "execution_plan",
        "artifact",
        "artifact_ref",
        "password",
        "secret",
        "credential",
        "credentials",
        "token",
        "api_key",
        "connection_string",
    }
)
_BINDING_SIDES = frozenset({"input", "output"})
_BINDING_FIELDS = frozenset(
    {
        "data_source_uid",
        "object_kind",
        "object_ref",
        "schema_snapshot_id",
        "schema_hash",
        "binding_hash",
        "access_mode",
        "write_mode",
        "dialect",
        "binding_id",
        "logical_ref",
    }
)


class DeploymentOperationError(ValueError, RuntimeError):
    code = "terminal_conflict"
    idempotency_key_disposition = "rotate"


class OperationInProgress(DeploymentOperationError):
    code = "operation_in_progress"
    idempotency_key_disposition = "retain"


class OperationUnknown(DeploymentOperationError):
    code = "operation_unknown"
    idempotency_key_disposition = "retain"

    def __init__(
        self,
        message: str,
        *,
        operation: Mapping[str, Any] | None = None,
    ) -> None:
        super().__init__(message)
        self.blocking_operation = (
            {
                "id": operation.get("id"),
                "deployment_id": operation.get("deployment_id"),
                "action": operation.get("action"),
                "idempotency_key": operation.get("idempotency_key"),
                "scope_unknown_count": int(operation.get("scope_unknown_count") or 1),
            }
            if operation is not None
            else None
        )


class TerminalConflict(DeploymentOperationError):
    code = "terminal_conflict"


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 _text(value: Any, label: str, limit: int) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{label} is required")
    normalized = value.strip()
    if len(normalized) > limit:
        raise ValueError(f"{label} exceeds {limit} characters")
    return normalized


def _digest(value: Any, label: str) -> str:
    normalized = str(value or "").strip().lower()
    if not re.fullmatch(r"[0-9a-f]{64}", normalized):
        raise ValueError(f"{label} must be a sha256 hex digest")
    return normalized


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


def _timestamp(value: Any, label: str) -> datetime:
    if isinstance(value, datetime):
        parsed = value
    elif isinstance(value, str):
        source = value[:-1] + "+00:00" if value.endswith("Z") else value
        try:
            parsed = datetime.fromisoformat(source)
        except ValueError as exc:
            raise ValueError(f"{label} is invalid") from exc
    else:
        raise ValueError(f"{label} is invalid")
    if parsed.tzinfo is None:
        raise ValueError(f"{label} must include a timezone")
    return parsed.astimezone(UTC)


def _closed_tree(value: Any) -> None:
    if isinstance(value, Mapping):
        for key, item in value.items():
            normalized = str(key).strip().lower()
            if normalized in _FORBIDDEN_INPUT_KEYS:
                if normalized in {
                    "password",
                    "secret",
                    "credential",
                    "credentials",
                    "token",
                    "api_key",
                    "connection_string",
                }:
                    raise ValueError("deployment credentials are not accepted")
                raise ValueError("inline executable semantics are not accepted")
            _closed_tree(item)
    elif isinstance(value, list):
        for item in value:
            _closed_tree(item)


def _binding_snapshot(value: Any) -> dict[str, Any]:
    if not isinstance(value, Mapping) or set(value) != _BINDING_SIDES:
        raise ValueError("binding_snapshot must contain input and output")
    result: dict[str, Any] = {}
    for side in ("input", "output"):
        raw = value[side]
        if not isinstance(raw, Mapping):
            raise ValueError(f"{side} binding must be an object")
        unknown = set(raw) - _BINDING_FIELDS
        if unknown:
            raise ValueError(
                f"{side} binding contains unsupported fields: "
                + ", ".join(sorted(unknown))
            )
        _closed_tree(raw)
        required = {
            "data_source_uid",
            "object_kind",
            "object_ref",
            "schema_snapshot_id",
            "schema_hash",
            "binding_hash",
            "access_mode",
            "dialect",
        }
        if not required.issubset(raw):
            raise ValueError(f"{side} binding is incomplete")
        item = {
            "data_source_uid": _uid(raw["data_source_uid"], f"{side} data_source_uid"),
            "object_kind": _text(raw["object_kind"], f"{side} object_kind", 40),
            "object_ref": _text(raw["object_ref"], f"{side} object_ref", 500),
            "schema_snapshot_id": _uid(
                raw["schema_snapshot_id"], f"{side} schema_snapshot_id"
            ),
            "schema_hash": _digest(raw["schema_hash"], f"{side} schema_hash"),
            "binding_hash": _digest(raw["binding_hash"], f"{side} binding_hash"),
            "access_mode": _text(raw["access_mode"], f"{side} access_mode", 20),
            "dialect": _text(raw["dialect"], f"{side} dialect", 40).lower(),
        }
        if raw.get("binding_id") is not None:
            item["binding_id"] = _uid(raw["binding_id"], f"{side} binding_id")
        if raw.get("logical_ref") is not None:
            item["logical_ref"] = _text(raw["logical_ref"], f"{side} logical_ref", 500)
        if "write_mode" in raw and raw["write_mode"] is not None:
            item["write_mode"] = _text(raw["write_mode"], f"{side} write_mode", 40)
        result[side] = item
    if result["input"]["access_mode"] != "read":
        raise ValueError("input binding access_mode must be read")
    if result["output"]["access_mode"] not in {"write", "read_write"}:
        raise ValueError("output binding access_mode must be write or read_write")
    if result["input"].get("write_mode"):
        raise ValueError("input binding cannot declare write_mode")
    if not result["output"].get("write_mode"):
        raise ValueError("output binding requires write_mode")
    return result


def _safe_receipt(value: Any) -> dict[str, Any]:
    if not isinstance(value, Mapping):
        return {}
    allowed = {"id", "revision", "status", "uid", "namespace", "flowId"}
    return {
        str(key): copy.deepcopy(item)
        for key, item in value.items()
        if key in allowed and isinstance(item, (str, int, float, bool, type(None)))
    }


class DataFlowDeploymentService:
    """Deploy one released production line through fail-closed transitions."""

    def __init__(
        self,
        repository,
        *,
        engine,
        compiler: Callable[..., Any] | None = None,
        token_issuer=None,
        clock: Callable[[], datetime] | None = None,
        canary_ttl_seconds: int = 900,
        canary_timeout_seconds: int = 60,
        poll_interval_seconds: float = 1.0,
        monotonic: Callable[[], float] | None = None,
        sleeper: Callable[[float], None] | None = None,
        operation_lease_seconds: int = 1200,
    ):
        if engine is None:
            raise RuntimeError("deployment engine is not configured")
        self.repository = repository
        self.engine = engine
        self.compiler = compiler or compile_kestra_flow
        self.token_issuer = token_issuer
        self.clock = clock or (lambda: datetime.now(UTC))
        self.monotonic = monotonic or time.monotonic
        self.sleeper = sleeper or time.sleep
        if operation_lease_seconds < 60 or operation_lease_seconds > 3600:
            raise ValueError("operation lease is outside policy")
        self.operation_lease_seconds = operation_lease_seconds
        if canary_ttl_seconds < 60 or canary_ttl_seconds > 3600:
            raise ValueError("canary evidence ttl is outside policy")
        self.canary_ttl_seconds = canary_ttl_seconds
        if canary_timeout_seconds < 1 or canary_timeout_seconds > 900:
            raise ValueError("canary timeout is outside policy")
        if poll_interval_seconds <= 0 or poll_interval_seconds > 10:
            raise ValueError("canary poll interval is outside policy")
        self.canary_timeout_seconds = canary_timeout_seconds
        self.poll_interval_seconds = poll_interval_seconds

    @staticmethod
    def _operation_key(value: Any) -> str:
        return _text(value, "idempotency_key", 200)

    @staticmethod
    def _correlation(value: Any) -> str:
        return _uid(value or new_governance_uid(), "correlation_id")

    def create(
        self,
        dataflow_version_id: str,
        *,
        binding_snapshot: dict[str, Any],
        environment: str,
        schedule_plan: dict[str, Any],
        actor_uid: str,
        reason: str,
        idempotency_key: str,
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        version_id = _uid(dataflow_version_id, "dataflow_version_id")
        actor = _uid(actor_uid, "actor_uid")
        if environment not in ENVIRONMENTS:
            raise ValueError("deployment environment is invalid")
        key = self._operation_key(idempotency_key)
        why = _text(reason, "reason", 1000)
        correlation = self._correlation(correlation_id)
        requested_binding = binding_snapshot
        schedule = validate_schedule_plan(schedule_plan)
        release = self.repository.load_deployable_release(
            version_id, environment=environment
        )
        if (
            not isinstance(release, dict)
            or release.get("status") != "released"
            or str(release.get("id")) != version_id
        ):
            raise ValueError("only a released dataflow version may be deployed")
        package = release.get("package")
        if not isinstance(package, dict):
            raise ValueError("released production-line package is missing")
        package_hash = _digest(release.get("package_hash"), "package_hash")
        if package.get("package_hash") != package_hash:
            raise ValueError("released production-line package has drifted")
        released_workflow_spec = package.get("workflow_spec")
        if not isinstance(released_workflow_spec, dict):
            raise ValueError("released workflow spec is missing")
        if released_workflow_spec.get("dataflow_uid") != package.get("dataflow_uid"):
            raise ValueError("released workflow identity has drifted")
        workflow_spec = release.get("physical_workflow_spec")
        if not isinstance(workflow_spec, dict):
            raise ValueError("server-owned physical workflow spec is required")
        if workflow_spec.get("dataflow_uid") != package.get("dataflow_uid"):
            raise ValueError("physical workflow identity has drifted")
        physical_hashes = sorted(
            {
                _digest(item, "physical_plan_hash")
                for item in release.get("physical_plan_hashes", [])
            }
        )
        if not physical_hashes:
            raise ValueError("published physical execution plans are required")
        schema_snapshots = release.get("schema_snapshots")
        if not isinstance(schema_snapshots, dict):
            raise ValueError("server-owned schema snapshots are required")
        canonical_binding = _binding_snapshot(release.get("binding_snapshot"))
        if (
            isinstance(requested_binding, Mapping)
            and set(requested_binding) == {"input", "output"}
            and all(
                isinstance(requested_binding[side], Mapping)
                and set(requested_binding[side]) == {"binding_id"}
                for side in ("input", "output")
            )
        ):
            if any(
                _uid(
                    requested_binding[side]["binding_id"],
                    f"{side} binding_id",
                )
                != canonical_binding[side].get("binding_id")
                for side in ("input", "output")
            ):
                raise ValueError("deployment binding selection has drifted")
        else:
            requested = _binding_snapshot(requested_binding)
            if requested != canonical_binding:
                raise ValueError("deployment binding selection has drifted")
        binding = canonical_binding
        for side in ("input", "output"):
            canonical = schema_snapshots.get(side)
            if (
                not isinstance(canonical, dict)
                or binding[side]["schema_snapshot_id"] != canonical.get("id")
                or binding[side]["schema_hash"] != canonical.get("schema_hash")
            ):
                raise ValueError("deployment binding schema has drifted")
        snapshot = {
            "dataflow_version_id": version_id,
            "version_no": int(release.get("version_no") or 0),
            "dataflow_uid": _uid(package.get("dataflow_uid"), "dataflow_uid"),
            "environment": environment,
            "package": copy.deepcopy(package),
            "package_hash": package_hash,
            "standard_version_ids": [
                _uid(item, "standard_version_id")
                for item in package.get("standard_version_ids", [])
            ],
            "rule_version_ids": [
                _uid(item, "rule_version_id")
                for item in package.get("rule_version_ids", [])
            ],
            "binding_snapshot": binding,
            "binding_hash": _canonical_hash(binding),
            "schema_snapshots": copy.deepcopy(schema_snapshots),
            "schema_snapshot_hash": _canonical_hash(schema_snapshots),
            "physical_plan_hashes": physical_hashes,
            "workflow_spec": copy.deepcopy(workflow_spec),
            "workflow_spec_hash": _canonical_hash(workflow_spec),
            "schedule_snapshot": schedule,
            "schedule_hash": _canonical_hash(schedule),
            "actor_uid": actor,
            "reason": why,
            "idempotency_key": key,
            "correlation_id": correlation,
            "preparatory_deployment_id": release.get("preparatory_deployment_id"),
        }
        return self.repository.create_deployment(snapshot)

    def _claim(
        self,
        deployment_id: str,
        action: str,
        actor_uid: str,
        idempotency_key: str,
        correlation_id: str | None,
        reason: str,
        request_payload: dict[str, Any],
    ):
        deployment = self.repository.get_deployment(
            _uid(deployment_id, "deployment_id")
        )
        actor = _uid(actor_uid, "actor_uid")
        key = self._operation_key(idempotency_key)
        correlation = self._correlation(correlation_id)
        claimed, operation = self.repository.claim_operation(
            deployment["id"],
            action,
            key,
            actor,
            correlation,
            reason,
            request_payload,
        )
        # Claim may also expire and classify an abandoned owner.  Commit both
        # successful claims and those durable state observations.
        self._commit()
        if not claimed:
            if operation.get("status") == "completed":
                return deployment, actor, operation, operation.get("result")
            if operation.get("status") == "claimed":
                raise OperationInProgress("deployment operation is already in progress")
            if operation.get("status") == "unknown":
                raise OperationUnknown(
                    "deployment operation outcome requires reconciliation",
                    operation=operation,
                )
            raise TerminalConflict(
                "failed deployment operation requires a new idempotency key"
            )
        return deployment, actor, operation, None

    def _commit(self) -> None:
        commit = getattr(self.repository, "commit", None)
        if commit is not None:
            commit()

    @staticmethod
    def _unknown_outcome(exc: Exception) -> bool:
        if isinstance(exc, (requests.Timeout, requests.ConnectionError)):
            return True
        if isinstance(exc, requests.HTTPError):
            response = getattr(exc, "response", None)
            return response is None or int(response.status_code) >= 500
        return False

    def _engine_failure(self, operation: Mapping[str, Any], exc: Exception):
        if self._unknown_outcome(exc):
            mark_unknown = getattr(self.repository, "mark_operation_unknown", None)
            if mark_unknown is not None:
                mark_unknown(operation, "deployment_engine_outcome_unknown")
            else:
                self.repository.fail_operation(
                    operation, "deployment_engine_outcome_unknown"
                )
            self._commit()
            raise OperationUnknown(
                "deployment engine outcome is unknown; reconciliation required"
            )
        self.repository.fail_operation(operation, "deployment_engine_failure")
        self._commit()
        raise RuntimeError("deployment engine operation failed")

    def _local_operation_failure(
        self, operation: Mapping[str, Any], error_code: str
    ) -> None:
        self.repository.fail_operation(operation, error_code)
        self._commit()

    def _owns_fence(self, operation: Mapping[str, Any]) -> bool:
        locker = getattr(self.repository, "lock_operation_lease", None)
        if locker is not None:
            return bool(locker(operation))
        checker = getattr(self.repository, "owns_operation_lease", None)
        return checker is None or bool(checker(operation))

    def _renew_fence(self, operation: Mapping[str, Any]) -> None:
        renew = getattr(self.repository, "renew_operation_lease", None)
        if renew is not None and not renew(operation, self.operation_lease_seconds):
            raise ValueError("deployment fencing lease lost")

    def _compiled_definition(self, deployment: Mapping[str, Any]):
        return self.compiler(
            deployment["workflow_spec"],
            deployment["schedule_snapshot"],
            deployment["environment"],
            int(deployment.get("version_no") or 1),
        )

    def _attest_compiled_flow(
        self,
        deployment: Mapping[str, Any],
        flow: Mapping[str, Any],
        compiled,
    ) -> str:
        labels = self._flow_labels(flow)
        expected_labels = {
            "dataflow_uid": deployment["dataflow_uid"],
            "environment": deployment["environment"],
            "workflow_version": str(int(deployment.get("version_no") or 1)),
        }
        if (
            flow.get("id") != compiled.flow_id
            or flow.get("namespace") != compiled.namespace
            or any(labels.get(key) != value for key, value in expected_labels.items())
        ):
            raise ValueError("engine flow identity attestation failed")
        actual_hash = canonical_flow_definition_hash(flow)
        expected = getattr(compiled, "definition", None)
        if (
            isinstance(expected, Mapping)
            and canonical_flow_definition_hash(expected) != actual_hash
        ):
            raise ValueError("engine definition drift detected")
        return actual_hash

    def _ensure_definition_attestation(
        self, deployment: Mapping[str, Any]
    ) -> dict[str, Any]:
        flow = self.engine.get_flow(
            deployment["engine_namespace"],
            deployment["engine_definition_id"],
        )
        if not isinstance(flow, Mapping):
            raise ValueError("engine definition is not observable")
        actual_hash = canonical_flow_definition_hash(flow)
        if deployment.get("engine_definition_hash_version") == KESTRA_HASH_VERSION:
            if actual_hash != deployment.get("engine_definition_hash"):
                raise ValueError("engine definition drift detected")
            return dict(deployment)
        compiled = self._compiled_definition(deployment)
        self._attest_compiled_flow(deployment, flow, compiled)
        re_attest = getattr(self.repository, "re_attest_engine_definition", None)
        if re_attest is None:
            raise ValueError("legacy engine definition requires re-attestation")
        updated = re_attest(
            deployment["id"],
            deployment["lock_version"],
            definition_hash=actual_hash,
            revision=flow.get("revision"),
        )
        if updated.get("engine_definition_hash_version") != KESTRA_HASH_VERSION:
            raise ValueError("engine definition re-attestation failed")
        return updated

    def _assert_definition(self, deployment: Mapping[str, Any]) -> Mapping[str, Any]:
        if deployment.get("engine_definition_hash_version") != KESTRA_HASH_VERSION:
            raise ValueError("legacy engine definition requires re-attestation")
        flow = self.engine.get_flow(
            deployment["engine_namespace"],
            deployment["engine_definition_id"],
        )
        if not isinstance(flow, Mapping):
            raise ValueError("engine definition is not observable")
        actual_hash = canonical_flow_definition_hash(flow)
        if actual_hash != deployment.get("engine_definition_hash"):
            raise ValueError("engine definition drift detected")
        return flow

    def _get_flow_optional(self, namespace: str, definition_id: str):
        try:
            return self.engine.get_flow(namespace, definition_id)
        except (KeyError, LookupError):
            return None
        except requests.HTTPError as exc:
            response = getattr(exc, "response", None)
            if response is not None and int(response.status_code) == 404:
                return None
            raise

    def deploy_disabled(
        self,
        deployment_id: str,
        actor_uid: str,
        *,
        idempotency_key: str,
        reason: str = "deploy disabled",
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        _text(reason, "reason", 1000)
        deployment, _actor, operation, replay = self._claim(
            deployment_id,
            "deploy_disabled",
            actor_uid,
            idempotency_key,
            correlation_id,
            reason,
            {},
        )
        if replay is not None:
            return replay
        try:
            if deployment["status"] not in {"draft", "failed"}:
                raise ValueError("deployment is not eligible for disabled deploy")
            compiled = self.compiler(
                deployment["workflow_spec"],
                deployment["schedule_snapshot"],
                deployment["environment"],
                int(deployment.get("version_no") or 1),
            )
        except Exception:
            self._local_operation_failure(operation, "deployment_preflight_failed")
            raise
        if compiled.definition_hash != _canonical_hash(
            # The compiler owns canonical serialized output; this comparison is
            # deliberately against its own immutable payload.
            compiled.yaml
        ):
            # Existing compiler hashes bytes, while canonical JSON hashing a
            # string includes quotes. Preserve its hash but still validate it.
            _digest(compiled.definition_hash, "engine_definition_hash")
        try:
            self._renew_fence(operation)
            response = self.engine.deploy_disabled(compiled.yaml)
            flow = self.engine.get_flow(compiled.namespace, compiled.flow_id)
            if not isinstance(flow, Mapping) or flow.get("disabled") is not True:
                raise RuntimeError("disabled engine definition is not observable")
            actual_hash = self._attest_compiled_flow(deployment, flow, compiled)
        except Exception as exc:
            self._engine_failure(operation, exc)
        engine = {
            "namespace": compiled.namespace,
            "definition_id": compiled.flow_id,
            "revision": flow.get("revision"),
            "definition_hash": actual_hash,
            "definition_hash_version": KESTRA_HASH_VERSION,
            "receipt": _safe_receipt(flow) or _safe_receipt(response),
        }
        try:
            result = self.repository.mark_disabled(
                deployment["id"],
                deployment["lock_version"],
                engine,
                operation,
            )
        except Exception:
            self._local_operation_failure(operation, "disabled_finalize_failed")
            raise
        self._commit()
        return result

    def run_canary(
        self,
        deployment_id: str,
        inputs: dict[str, Any],
        actor_uid: str,
        *,
        idempotency_key: str,
        reason: str = "trial production",
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        _text(reason, "reason", 1000)
        if not isinstance(inputs, dict) or len(inputs) > 100:
            raise ValueError("canary inputs must be a bounded object")
        _closed_tree(inputs)
        if len(json.dumps(inputs, ensure_ascii=False).encode("utf-8")) > 65536:
            raise ValueError("canary inputs exceed size limit")
        deployment, actor, operation, replay = self._claim(
            deployment_id,
            "run_canary",
            actor_uid,
            idempotency_key,
            correlation_id,
            reason,
            {"inputs_hash": _canonical_hash(inputs)},
        )
        if replay is not None:
            return replay
        try:
            if deployment["status"] not in {"disabled", "canary"}:
                raise ValueError("disabled deployment is required for canary")
            deployment = self._ensure_definition_attestation(deployment)
            allowed_inputs = set(
                (deployment["workflow_spec"].get("parameters") or {}).keys()
            )
            unknown = sorted(set(inputs) - allowed_inputs)
            if unknown:
                raise ValueError(
                    "canary inputs contain unknown parameters: " + ", ".join(unknown)
                )
            engine_inputs = copy.deepcopy(inputs)
            if self.token_issuer is not None:
                task_tokens = {}
                for node in deployment["workflow_spec"].get("nodes", []):
                    task_tokens[node["id"]] = self.token_issuer.issue(
                        task_uid=new_governance_uid(),
                        dataflow_uid=deployment["dataflow_uid"],
                        deployment_id=deployment["id"],
                        environment=deployment["environment"],
                        workflow_version=int(deployment.get("version_no") or 1),
                        correlation_id=operation["correlation_id"],
                        node=node,
                        write_authorized=node.get("purpose") == "write",
                    )
                engine_inputs["dataops_task_tokens"] = task_tokens
        except Exception:
            self._local_operation_failure(operation, "canary_preflight_failed")
            raise
        try:
            self._renew_fence(operation)
            self.engine.activate(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
            try:
                response = self.engine.execute(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                    inputs=engine_inputs,
                )
            finally:
                self._renew_fence(operation)
                self.engine.deactivate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
            execution_id = response.get("id") if isinstance(response, Mapping) else None
            if not execution_id:
                raise RuntimeError("canary execution id is missing")
            deadline = self.monotonic() + self.canary_timeout_seconds
            execution = self.engine.get_execution(str(execution_id))
            while True:
                state = (
                    execution.get("state") if isinstance(execution, Mapping) else None
                )
                if isinstance(state, Mapping):
                    state = state.get("current")
                normalized_state = str(state or "").upper()
                if normalized_state in {
                    "SUCCESS",
                    "SUCCEEDED",
                    "FAILED",
                    "KILLED",
                    "CANCELLED",
                    "WARNING",
                }:
                    break
                if self.monotonic() >= deadline:
                    normalized_state = "UNKNOWN"
                    break
                self.sleeper(self.poll_interval_seconds)
                self._renew_fence(operation)
                execution = self.engine.get_execution(str(execution_id))
        except Exception as exc:
            self._engine_failure(operation, exc)
        passed = normalized_state in {"SUCCESS", "SUCCEEDED"}
        evidence_status = (
            "passed"
            if passed
            else (
                "failed"
                if normalized_state in {"FAILED", "KILLED", "CANCELLED", "WARNING"}
                else "unknown"
            )
        )
        now = self.clock().astimezone(UTC)
        evidence = {
            "id": new_governance_uid(),
            "deployment_id": deployment["id"],
            "execution_id": str(execution_id),
            "status": evidence_status,
            "package_hash": deployment["package_hash"],
            "binding_hash": deployment["binding_hash"],
            "schema_snapshot_hash": deployment["schema_snapshot_hash"],
            "physical_plan_hashes": deployment["physical_plan_hashes"],
            "workflow_spec_hash": deployment["workflow_spec_hash"],
            "schedule_hash": deployment["schedule_hash"],
            "engine_definition_hash": deployment["engine_definition_hash"],
            "engine_definition_hash_version": KESTRA_HASH_VERSION,
            "verified_by": actor,
            "created_at": now.isoformat(),
            "expires_at": (
                now + timedelta(seconds=self.canary_ttl_seconds)
            ).isoformat(),
            "engine_response": _safe_receipt(execution),
        }
        try:
            result = self.repository.record_canary(
                deployment["id"],
                deployment["lock_version"],
                evidence,
                operation,
            )
        except Exception:
            self._local_operation_failure(operation, "canary_finalize_failed")
            raise
        self._commit()
        return result

    def _validate_evidence(
        self, deployment: Mapping[str, Any], evidence_id: Any
    ) -> dict[str, Any]:
        if not evidence_id:
            raise ValueError("passed canary evidence is required")
        evidence = self.repository.get_canary_evidence(_uid(evidence_id, "evidence_id"))
        if evidence.get("status") != "passed" or not evidence.get("verified_by"):
            raise ValueError("passed canary evidence is required")
        expires_at = _timestamp(evidence.get("expires_at"), "canary expiry")
        if expires_at <= self.clock().astimezone(UTC):
            raise ValueError("passed canary evidence has expired")
        expected = {
            "deployment_id": deployment["id"],
            "package_hash": deployment["package_hash"],
            "binding_hash": deployment["binding_hash"],
            "schema_snapshot_hash": deployment["schema_snapshot_hash"],
            "physical_plan_hashes": deployment["physical_plan_hashes"],
            "workflow_spec_hash": deployment["workflow_spec_hash"],
            "schedule_hash": deployment["schedule_hash"],
            "engine_definition_hash": deployment["engine_definition_hash"],
            "engine_definition_hash_version": KESTRA_HASH_VERSION,
        }
        if any(evidence.get(key) != value for key, value in expected.items()):
            raise ValueError("canary evidence attestation has drifted")
        return evidence

    def execute_active(
        self,
        deployment_id: str,
        inputs: dict[str, Any],
        actor_uid: str,
        *,
        idempotency_key: str,
        reason: str = "manual production execution",
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        """Launch one active production line with server-issued task tokens."""

        _text(reason, "reason", 1000)
        if not isinstance(inputs, dict) or len(inputs) > 100:
            raise ValueError("production inputs must be a bounded object")
        _closed_tree(inputs)
        if len(json.dumps(inputs, ensure_ascii=False).encode("utf-8")) > 65536:
            raise ValueError("production inputs exceed size limit")
        deployment, _actor, operation, replay = self._claim(
            deployment_id,
            "execute_active",
            actor_uid,
            idempotency_key,
            correlation_id,
            reason,
            {"inputs_hash": _canonical_hash(inputs)},
        )
        if replay is not None:
            return replay
        try:
            if deployment["status"] != "active":
                raise ValueError("active deployment is required for execution")
            deployment = self._ensure_definition_attestation(deployment)
            allowed_inputs = set(
                (deployment["workflow_spec"].get("parameters") or {}).keys()
            )
            unknown = sorted(set(inputs) - allowed_inputs)
            if unknown:
                raise ValueError(
                    "production inputs contain unknown parameters: "
                    + ", ".join(unknown)
                )
            if self.token_issuer is None:
                raise RuntimeError("production task token issuer is not configured")
            task_tokens = {}
            for node in deployment["workflow_spec"].get("nodes", []):
                task_tokens[node["id"]] = self.token_issuer.issue(
                    task_uid=new_governance_uid(),
                    dataflow_uid=deployment["dataflow_uid"],
                    deployment_id=deployment["id"],
                    environment=deployment["environment"],
                    workflow_version=int(deployment.get("version_no") or 1),
                    correlation_id=operation["correlation_id"],
                    node=node,
                    write_authorized=node.get("purpose") == "write",
                )
            engine_inputs = {
                **copy.deepcopy(inputs),
                "dataops_task_tokens": task_tokens,
            }
        except Exception:
            self._local_operation_failure(
                operation, "active_execution_preflight_failed"
            )
            raise
        try:
            self._renew_fence(operation)
            response = self.engine.execute(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
                inputs=engine_inputs,
            )
            execution_id = response.get("id") if isinstance(response, Mapping) else None
            if not execution_id:
                raise RuntimeError("production execution id is missing")
            receipt = {
                "deployment_id": deployment["id"],
                "dataflow_uid": deployment["dataflow_uid"],
                "dataflow_version_id": deployment["dataflow_version_id"],
                "workflow_version": int(deployment.get("version_no") or 1),
                "execution_id": str(execution_id),
                "correlation_id": operation["correlation_id"],
                "status": "submitted",
                "engine_response": _safe_receipt(response),
            }
            self.repository.record_active_execution_receipt(operation, receipt)
            # Recovery boundary: after this commit, reconciliation observes the
            # same execution id and must never resubmit the workflow.
            self._commit()
        except Exception as exc:
            self._engine_failure(operation, exc)
        try:
            deadline = self.monotonic() + self.canary_timeout_seconds
            execution = self.engine.get_execution(str(execution_id))
            while True:
                state = (
                    execution.get("state") if isinstance(execution, Mapping) else None
                )
                if isinstance(state, Mapping):
                    state = state.get("current")
                normalized_state = str(state or "").upper()
                if normalized_state in {
                    "SUCCESS",
                    "SUCCEEDED",
                    "FAILED",
                    "KILLED",
                    "CANCELLED",
                    "WARNING",
                }:
                    break
                if self.monotonic() >= deadline:
                    normalized_state = "UNKNOWN"
                    break
                self.sleeper(self.poll_interval_seconds)
                self._renew_fence(operation)
                execution = self.engine.get_execution(str(execution_id))
        except Exception as exc:
            self._engine_failure(operation, exc)
        if normalized_state not in {"SUCCESS", "SUCCEEDED"}:
            failure = (
                "active_execution_unknown"
                if normalized_state == "UNKNOWN"
                else "active_execution_failed"
            )
            if failure == "active_execution_unknown":
                self.repository.mark_operation_unknown(operation, failure)
                self._commit()
                raise OperationUnknown(
                    "production execution outcome requires reconciliation",
                    operation=operation,
                )
            self.repository.fail_operation(operation, failure)
            self._commit()
            raise RuntimeError("production execution failed")
        result = {
            "deployment_id": deployment["id"],
            "dataflow_uid": deployment["dataflow_uid"],
            "dataflow_version_id": deployment["dataflow_version_id"],
            "workflow_version": int(deployment.get("version_no") or 1),
            "execution_id": str(execution_id),
            "correlation_id": operation["correlation_id"],
            "status": "success",
            "engine_response": _safe_receipt(execution),
        }
        try:
            self.repository.complete_active_execution(operation, result)
        except Exception:
            self._local_operation_failure(operation, "active_execution_finalize_failed")
            raise
        self._commit()
        return result

    def activate(
        self,
        deployment_id: str,
        evidence_id: str | None,
        actor_uid: str,
        *,
        idempotency_key: str,
        reason: str = "mass production activation",
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        _text(reason, "reason", 1000)
        deployment, actor, operation, replay = self._claim(
            deployment_id,
            "activate",
            actor_uid,
            idempotency_key,
            correlation_id,
            reason,
            {"evidence_id": evidence_id},
        )
        if replay is not None:
            return replay
        try:
            if deployment["status"] != "canary":
                raise ValueError("passed canary deployment is required")
            deployment = self._ensure_definition_attestation(deployment)
            evidence = self._validate_evidence(deployment, evidence_id)
        except Exception:
            self._local_operation_failure(operation, "activation_preflight_failed")
            raise
        try:
            self._renew_fence(operation)
            self.engine.activate(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
        except Exception as exc:
            self._engine_failure(operation, exc)
        try:
            result = self.repository.activate_atomic(
                deployment["id"],
                deployment["lock_version"],
                evidence["id"],
                actor,
                operation,
            )
            self._commit()
            return result
        except Exception:
            if self._owns_fence(operation):
                try:
                    self.engine.deactivate(
                        deployment["engine_namespace"],
                        deployment["engine_definition_id"],
                    )
                finally:
                    self.repository.fail_operation(
                        operation, "activation_commit_failure"
                    )
                    self._commit()
            raise

    def rollback(
        self,
        deployment_id: str,
        actor_uid: str,
        *,
        idempotency_key: str,
        reason: str = "restore previous production line",
        correlation_id: str | None = None,
    ) -> dict[str, Any]:
        _text(reason, "reason", 1000)
        deployment, actor, operation, replay = self._claim(
            deployment_id,
            "rollback",
            actor_uid,
            idempotency_key,
            correlation_id,
            reason,
            {
                "candidate_deployment_id": deployment_id,
            },
        )
        if replay is not None:
            return replay
        try:
            if deployment["status"] not in {"active", "failed"}:
                raise ValueError("active deployment is required for rollback")
            previous_id = deployment.get("previous_active_deployment_id")
            if not previous_id:
                raise ValueError("previous deployment is not available")
            previous = self.repository.get_deployment(previous_id)
            if (
                previous.get("status") != "superseded"
                or previous.get("dataflow_uid") != deployment.get("dataflow_uid")
                or previous.get("environment") != deployment.get("environment")
            ):
                raise ValueError("previous deployment snapshot is not restorable")
            deployment = self._ensure_definition_attestation(deployment)
            previous = self._ensure_definition_attestation(previous)
        except Exception:
            self._local_operation_failure(operation, "rollback_preflight_failed")
            raise
        try:
            # Fail-safe ordering: stop the candidate before restoring the exact
            # prior engine definition and immutable package.
            self._renew_fence(operation)
            self.engine.deactivate(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
            self._renew_fence(operation)
            self.engine.activate(
                previous["engine_namespace"],
                previous["engine_definition_id"],
            )
        except Exception as exc:
            if self._unknown_outcome(exc):
                self._engine_failure(operation, exc)
            if self._owns_fence(operation):
                try:
                    self.engine.activate(
                        deployment["engine_namespace"],
                        deployment["engine_definition_id"],
                    )
                finally:
                    self._engine_failure(operation, exc)
            raise
        try:
            result = self.repository.rollback_atomic(
                deployment["id"],
                deployment["lock_version"],
                actor,
                operation,
            )
            self._commit()
            return result
        except Exception:
            # External state is reconciled back to the pre-operation state.
            if self._owns_fence(operation):
                self.engine.deactivate(
                    previous["engine_namespace"], previous["engine_definition_id"]
                )
                self.engine.activate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
                self.repository.fail_operation(operation, "rollback_commit_failure")
                self._commit()
            raise

    @staticmethod
    def _flow_labels(flow: Mapping[str, Any]) -> dict[str, str]:
        labels = flow.get("labels")
        if isinstance(labels, Mapping):
            return {str(key): str(value) for key, value in labels.items()}
        if isinstance(labels, list):
            return {
                str(item.get("key")): str(item.get("value"))
                for item in labels
                if isinstance(item, Mapping)
                and item.get("key") is not None
                and item.get("value") is not None
            }
        return {}

    def reconcile(
        self,
        deployment_id: str,
        action: str,
        actor_uid: str,
        *,
        idempotency_key: str,
    ) -> dict[str, Any]:
        """Resolve a response-loss outcome without replaying the side effect."""

        deployment = self.repository.get_deployment(
            _uid(deployment_id, "deployment_id")
        )
        actor = _uid(actor_uid, "actor_uid")
        operation = self.repository.get_operation(
            deployment["id"], action, self._operation_key(idempotency_key)
        )
        if operation.get("status") == "completed":
            return operation["result"]
        if operation.get("actor_uid") != actor:
            raise ValueError("operation is bound to another actor")
        if operation.get("status") == "claimed":
            expires_at = operation.get("lease_expires_at")
            if expires_at and _timestamp(expires_at, "operation lease") > self.clock():
                raise OperationInProgress("deployment operation is already in progress")
        if operation.get("status") not in {"claimed", "unknown"}:
            raise TerminalConflict("terminal deployment operation cannot reconcile")
        recover = getattr(self.repository, "recover_operation", None)
        if recover is not None:
            try:
                operation = recover(
                    deployment["id"],
                    action,
                    self._operation_key(idempotency_key),
                    actor,
                )
            except ValueError as exc:
                if "already in progress" in str(exc):
                    raise OperationInProgress(str(exc)) from exc
                raise
        else:
            self.repository.reclaim_unknown_operation(operation["id"])
            operation = self.repository.get_operation(
                deployment["id"], action, self._operation_key(idempotency_key)
            )
        self._commit()
        if action == "execute_active":
            receipt = operation.get("result")
            if not isinstance(receipt, Mapping) or not receipt.get("execution_id"):
                self.repository.mark_operation_unknown(
                    operation, "active_execution_receipt_missing"
                )
                self._commit()
                raise OperationUnknown(
                    "production execution receipt is missing; "
                    "the execution will not be replayed",
                    operation=operation,
                )
            try:
                execution = self.engine.get_execution(
                    str(receipt["execution_id"])
                )
            except Exception as exc:
                self.repository.mark_operation_unknown(
                    operation, "active_execution_reconcile_unavailable"
                )
                self._commit()
                raise OperationUnknown(
                    "production execution state is unavailable",
                    operation=operation,
                ) from exc
            state = (
                execution.get("state")
                if isinstance(execution, Mapping)
                else None
            )
            if isinstance(state, Mapping):
                state = state.get("current")
            normalized_state = str(state or "").upper()
            if normalized_state in {"SUCCESS", "SUCCEEDED"}:
                result = {
                    **dict(receipt),
                    "status": "success",
                    "engine_response": _safe_receipt(execution),
                }
                self.repository.complete_active_execution(operation, result)
                self._commit()
                return result
            if normalized_state in {
                "FAILED",
                "KILLED",
                "CANCELLED",
                "WARNING",
            }:
                self.repository.fail_operation(
                    operation, "active_execution_failed"
                )
                self._commit()
                raise RuntimeError("production execution failed")
            self.repository.mark_operation_unknown(
                operation, "active_execution_not_terminal"
            )
            self._commit()
            raise OperationUnknown(
                "production execution is not terminal",
                operation=operation,
            )
        if action == "deploy_disabled":
            compiled = self._compiled_definition(deployment)
            flow = self._get_flow_optional(compiled.namespace, compiled.flow_id)
            receipt = {}
            if flow is None:
                self._renew_fence(operation)
                receipt = self.engine.deploy_disabled(compiled.yaml)
                flow = self.engine.get_flow(compiled.namespace, compiled.flow_id)
            if not isinstance(flow, Mapping) or flow.get("disabled") is not True:
                raise ValueError("engine flow does not attest the unknown deployment")
            actual_hash = self._attest_compiled_flow(deployment, flow, compiled)
            result = self.repository.mark_disabled(
                deployment["id"],
                deployment["lock_version"],
                {
                    "namespace": compiled.namespace,
                    "definition_id": compiled.flow_id,
                    "revision": flow.get("revision"),
                    "definition_hash": actual_hash,
                    "definition_hash_version": KESTRA_HASH_VERSION,
                    "receipt": _safe_receipt(flow) or _safe_receipt(receipt),
                },
                operation,
            )
            self._commit()
            return result
        if action == "activate":
            evidence_id = operation.get("request", {}).get("evidence_id")
            deployment = self._ensure_definition_attestation(deployment)
            evidence = self._validate_evidence(deployment, evidence_id)
            flow = self._assert_definition(deployment)
            if flow.get("disabled") is True:
                self._renew_fence(operation)
                self.engine.activate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
                flow = self._assert_definition(deployment)
            if flow.get("disabled") is not False:
                raise ValueError("engine activation is not observable")
            result = self.repository.activate_atomic(
                deployment["id"],
                deployment["lock_version"],
                evidence["id"],
                actor,
                operation,
            )
            self._commit()
            return result
        if action == "rollback":
            if deployment.get("status") not in {
                "active",
                "failed",
            } or not deployment.get("previous_active_deployment_id"):
                raise ValueError("rollback deployment state is not recoverable")
            previous = self.repository.get_deployment(
                deployment["previous_active_deployment_id"]
            )
            if (
                previous.get("status") != "superseded"
                or previous.get("dataflow_uid") != deployment.get("dataflow_uid")
                or previous.get("environment") != deployment.get("environment")
            ):
                raise ValueError("previous deployment is not restorable")
            deployment = self._ensure_definition_attestation(deployment)
            previous = self._ensure_definition_attestation(previous)
            candidate_flow = self._assert_definition(deployment)
            previous_flow = self._assert_definition(previous)
            state = (
                candidate_flow.get("disabled"),
                previous_flow.get("disabled"),
            )
            if state == (False, True):
                self._renew_fence(operation)
                self.engine.deactivate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
                state = (True, True)
            if state == (True, True):
                self._renew_fence(operation)
                self.engine.activate(
                    previous["engine_namespace"],
                    previous["engine_definition_id"],
                )
                state = (True, False)
            if state != (True, False):
                raise ValueError("engine rollback state is not safely recoverable")
            candidate_flow = self._assert_definition(deployment)
            previous_flow = self._assert_definition(previous)
            if (
                candidate_flow.get("disabled") is not True
                or previous_flow.get("disabled") is not False
            ):
                raise ValueError("engine rollback state is not observable")
            result = self.repository.rollback_atomic(
                deployment["id"],
                deployment["lock_version"],
                actor,
                operation,
            )
            self._commit()
            return result
        raise ValueError(
            "unknown operation requires execution-level manual reconciliation"
        )

    def list(self, *, environment: str | None = None):
        if environment is not None and environment not in ENVIRONMENTS:
            raise ValueError("deployment environment is invalid")
        return self.repository.list_deployments(environment=environment)

    def candidates(self, *, environment: str):
        if environment not in ENVIRONMENTS:
            raise ValueError("deployment environment is invalid")
        return self.repository.list_deployable_releases(environment=environment)