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.spec import validate_schedule_plan

ENVIRONMENTS = frozenset({"development", "test", "production"})
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",
    }
)


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"] != "write":
        raise ValueError("output binding access_mode must be 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,
    ):
        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 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")
        workflow_spec = package.get("workflow_spec")
        if not isinstance(workflow_spec, dict):
            raise ValueError("released workflow spec is missing")
        if workflow_spec.get("dataflow_uid") != package.get("dataflow_uid"):
            raise ValueError("released 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,
        )
        if claimed:
            self._commit()
        if not claimed:
            if operation.get("status") == "completed":
                return deployment, actor, operation, operation.get("result")
            if operation.get("status") == "claimed":
                raise ValueError("deployment operation is already in progress")
            if operation.get("status") == "unknown":
                raise ValueError(
                    "deployment operation outcome requires reconciliation"
                )
            raise ValueError(
                "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["id"], "deployment_engine_outcome_unknown"
                )
            else:
                self.repository.fail_operation(
                    operation["id"], "deployment_engine_outcome_unknown"
                )
            self._commit()
            raise RuntimeError(
                "deployment engine outcome is unknown; reconciliation required"
            )
        self.repository.fail_operation(
            operation["id"], "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["id"], error_code)
        self._commit()

    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:
            response = self.engine.deploy_disabled(compiled.yaml)
        except Exception as exc:
            self._engine_failure(operation, exc)
        engine = {
            "namespace": compiled.namespace,
            "definition_id": compiled.flow_id,
            "revision": (
                response.get("revision")
                if isinstance(response, Mapping)
                else None
            ),
            "definition_hash": _digest(
                compiled.definition_hash, "engine_definition_hash"
            ),
            "receipt": _safe_receipt(response),
        }
        try:
            result = self.repository.mark_disabled(
                deployment["id"],
                deployment["lock_version"],
                engine,
                operation["id"],
            )
        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")
            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=True,
                    )
                engine_inputs["dataops_task_tokens"] = task_tokens
        except Exception:
            self._local_operation_failure(
                operation, "canary_preflight_failed"
            )
            raise
        try:
            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.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)
                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"],
            "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["id"],
            )
        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"],
        }
        if any(evidence.get(key) != value for key, value in expected.items()):
            raise ValueError("canary evidence attestation has drifted")
        return evidence

    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")
            evidence = self._validate_evidence(deployment, evidence_id)
        except Exception:
            self._local_operation_failure(
                operation, "activation_preflight_failed"
            )
            raise
        try:
            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["id"],
            )
            self._commit()
            return result
        except Exception:
            try:
                self.engine.deactivate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
            finally:
                self.repository.fail_operation(
                    operation["id"], "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"
                )
        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.engine.deactivate(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
            self.engine.activate(
                previous["engine_namespace"],
                previous["engine_definition_id"],
            )
        except Exception as exc:
            try:
                self.engine.activate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
            finally:
                self._engine_failure(operation, exc)
        try:
            result = self.repository.rollback_atomic(
                deployment["id"],
                deployment["lock_version"],
                actor,
                operation["id"],
            )
            self._commit()
            return result
        except Exception:
            # External state is reconciled back to the pre-operation state.
            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["id"], "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("status") != "unknown":
            raise ValueError("only an unknown operation may be reconciled")
        if operation.get("actor_uid") != actor:
            raise ValueError("operation is bound to another actor")
        if action == "deploy_disabled":
            compiled = self.compiler(
                deployment["workflow_spec"],
                deployment["schedule_snapshot"],
                deployment["environment"],
                int(deployment.get("version_no") or 1),
            )
            flow = self.engine.get_flow(compiled.namespace, compiled.flow_id)
            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 (
                not isinstance(flow, Mapping)
                or flow.get("disabled") is not True
                or any(
                    labels.get(key) != value
                    for key, value in expected_labels.items()
                )
            ):
                raise ValueError(
                    "engine flow does not attest the unknown deployment"
                )
            self.repository.reclaim_unknown_operation(operation["id"])
            self._commit()
            result = self.repository.mark_disabled(
                deployment["id"],
                deployment["lock_version"],
                {
                    "namespace": compiled.namespace,
                    "definition_id": compiled.flow_id,
                    "revision": flow.get("revision"),
                    "definition_hash": compiled.definition_hash,
                    "receipt": _safe_receipt(flow),
                },
                operation["id"],
            )
            self._commit()
            return result
        if action == "activate":
            evidence_id = operation.get("request", {}).get("evidence_id")
            evidence = self._validate_evidence(deployment, evidence_id)
            flow = self.engine.get_flow(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
            if not isinstance(flow, Mapping) or flow.get("disabled") is not False:
                raise ValueError("engine activation is not observable")
            self.repository.reclaim_unknown_operation(operation["id"])
            self._commit()
            result = self.repository.activate_atomic(
                deployment["id"],
                deployment["lock_version"],
                evidence["id"],
                actor,
                operation["id"],
            )
            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")
            candidate_flow = self.engine.get_flow(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
            previous_flow = self.engine.get_flow(
                previous["engine_namespace"],
                previous["engine_definition_id"],
            )
            if candidate_flow.get("disabled") is not True:
                self.engine.deactivate(
                    deployment["engine_namespace"],
                    deployment["engine_definition_id"],
                )
            if previous_flow.get("disabled") is not False:
                self.engine.activate(
                    previous["engine_namespace"],
                    previous["engine_definition_id"],
                )
            candidate_flow = self.engine.get_flow(
                deployment["engine_namespace"],
                deployment["engine_definition_id"],
            )
            previous_flow = self.engine.get_flow(
                previous["engine_namespace"],
                previous["engine_definition_id"],
            )
            if (
                candidate_flow.get("disabled") is not True
                or previous_flow.get("disabled") is not False
            ):
                raise ValueError("engine rollback state is not observable")
            self.repository.reclaim_unknown_operation(operation["id"])
            self._commit()
            result = self.repository.rollback_atomic(
                deployment["id"],
                deployment["lock_version"],
                actor,
                operation["id"],
            )
            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
        )