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