from __future__ import annotations

from collections.abc import Mapping
import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timedelta

from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.automation import AutomationInvocation
from govoplan_core.core.dataflows import (
    DataflowPublicationTarget,
    DataflowRunConflictError,
    DataflowRunDescriptor,
    DataflowRunError,
    DataflowRunNotFoundError,
    DataflowRunRequest,
)
from govoplan_core.core.datasources import (
    DatasourceError,
    DatasourcePublicationRequest,
    DatasourceReadRequest,
    DatasourceUnavailableError,
    datasource_catalogue,
    datasource_publication,
)
from govoplan_core.db.base import utcnow
from govoplan_dataflow.backend.backends import (
    BackendExecutionError,
    BackendSource,
    ExecutionBudget,
    execute_typed_graph,
)
from govoplan_dataflow.backend.batches import TypedBatch
from govoplan_dataflow.backend.db.models import (
    DataflowPipeline,
    DataflowPipelineDeployment,
    DataflowPipelineRevision,
    DataflowRun,
    new_uuid,
)
from govoplan_dataflow.backend.executor import (
    EXECUTOR_VERSION,
    MAX_SOURCE_ROWS,
    PipelineExecutionError,
    PipelineExecutionResult,
    ResolvedSource,
    execute_preview,
)
from govoplan_dataflow.backend.governance import (
    definition_governance_payload,
    require_definition_action,
)
from govoplan_dataflow.backend.graph import (
    canonical_graph_payload,
    definition_hash,
    preserve_compatible_graph_layout,
    validate_graph,
)
from govoplan_dataflow.backend.ir import graph_to_ir
from govoplan_dataflow.backend.schemas import (
    DataflowDiagnostic,
    GraphNode,
    NodePreviewResult,
    PipelineCreateRequest,
    PipelineDeriveRequest,
    PipelineDraftRequest,
    PipelineGraph,
    PipelinePreviewRequest,
    PipelinePreviewResponse,
    PipelineDeploymentResponse,
    PipelinePromotionRequest,
    PipelineRebaseRequest,
    PipelineResponse,
    PipelineRevisionResponse,
    PipelineRunResponse,
    PipelineSqlResponse,
    PipelineUpdateRequest,
    PipelineValidationResponse,
    PreviewColumn,
)
from govoplan_dataflow.backend.recovery import (
    DataflowRecoveryError,
    DataflowRunRecovery,
    begin_dataflow_run_recovery,
    dataflow_run_recovery_state,
)
from govoplan_dataflow.backend.reconciliation_decisions import (
    current_decision_rows,
    decision_set_fingerprint,
    decision_set_id_from_ref,
    get_decision_set,
)
from govoplan_dataflow.backend.sql_compiler import (
    SqlCompilationError,
    compile_sql,
    render_sql,
)


class DataflowError(DataflowRunError):
    pass


class DataflowNotFoundError(DataflowError, DataflowRunNotFoundError):
    pass


class DataflowConflictError(DataflowError, DataflowRunConflictError):
    pass


class DataflowValidationError(DataflowError):
    def __init__(self, diagnostics: list[DataflowDiagnostic]) -> None:
        super().__init__(diagnostics[0].message if diagnostics else "Pipeline validation failed")
        self.diagnostics = diagnostics


MAX_PENDING_RUNS_PER_TENANT = 100
MAX_PRODUCTION_ROWS = 10_000
RUN_SCOPE = "dataflow:pipeline:run"
DATASOURCE_READ_SCOPE = "datasources:catalogue:read"
DATASOURCE_WRITE_SCOPE = "datasources:source:write"


@dataclass(frozen=True)
class NormalizedDefinition:
    graph: PipelineGraph
    sql_text: str | None
    diagnostics: list[DataflowDiagnostic]


def list_pipelines(session: Session, *, tenant_id: str) -> list[DataflowPipeline]:
    return list(
        session.scalars(
            select(DataflowPipeline)
            .where(
                or_(
                    DataflowPipeline.tenant_id == tenant_id,
                    DataflowPipeline.tenant_id.is_(None),
                ),
                DataflowPipeline.deleted_at.is_(None),
            )
            .order_by(DataflowPipeline.updated_at.desc(), DataflowPipeline.name)
        )
    )


def get_pipeline(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
) -> DataflowPipeline:
    pipeline = session.scalar(
        select(DataflowPipeline).where(
            DataflowPipeline.id == pipeline_id,
            or_(
                DataflowPipeline.tenant_id == tenant_id,
                DataflowPipeline.tenant_id.is_(None),
            ),
            DataflowPipeline.deleted_at.is_(None),
        )
    )
    if pipeline is None:
        raise DataflowNotFoundError("Dataflow pipeline not found")
    return pipeline


def get_pipeline_revision(
    session: Session,
    *,
    pipeline: DataflowPipeline,
    revision: int | None = None,
) -> DataflowPipelineRevision:
    revision_number = revision or pipeline.current_revision
    item = session.scalar(
        select(DataflowPipelineRevision).where(
            DataflowPipelineRevision.pipeline_id == pipeline.id,
            DataflowPipelineRevision.tenant_id == pipeline.tenant_id,
            DataflowPipelineRevision.revision == revision_number,
        )
    )
    if item is None:
        raise DataflowNotFoundError("Dataflow pipeline revision not found")
    return item


def _resolve_reusable_subflows(
    session: Session,
    *,
    tenant_id: str,
    graph: PipelineGraph,
    principal: ApiPrincipal | None,
    registry: object | None,
    target_pipeline_id: str | None,
    ancestry: tuple[str, ...] = (),
) -> PipelineGraph:
    if not any(node.type == "subflow" for node in graph.nodes):
        return graph
    if principal is None:
        raise DataflowConflictError(
            "Reusable subflows require a tenant principal and current Policy "
            "decision."
        )
    resolved_nodes: list[GraphNode] = []
    for node in graph.nodes:
        if node.type != "subflow":
            resolved_nodes.append(node)
            continue
        source_id = _pipeline_id_from_ref(node.config.get("template_ref"))
        if source_id == target_pipeline_id:
            raise DataflowConflictError(
                "A pipeline cannot reference itself as a reusable subflow."
            )
        source = get_pipeline(
            session,
            tenant_id=tenant_id,
            pipeline_id=source_id,
        )
        reuse_decision = require_definition_action(
            source,
            principal=principal,
            registry=registry,
            action="reuse",
        )
        source_revision_number = _subflow_revision(
            node.config.get("template_version")
        )
        source_revision = get_pipeline_revision(
            session,
            pipeline=source,
            revision=source_revision_number,
        )
        reference_key = f"{source.id}:{source_revision.revision}"
        if reference_key in ancestry:
            raise DataflowConflictError(
                "Reusable subflow references contain a cycle at "
                f"pipeline:{source.id} revision {source_revision.revision}."
            )
        nested = _resolve_reusable_subflows(
            session,
            tenant_id=tenant_id,
            graph=PipelineGraph.model_validate(source_revision.graph),
            principal=principal,
            registry=registry,
            target_pipeline_id=target_pipeline_id,
            ancestry=(*ancestry, reference_key),
        )
        input_nodes = [
            item
            for item in nested.nodes
            if item.type == "source.inline"
            and item.config.get("input_binding") is True
        ]
        if len(input_nodes) != 1:
            raise DataflowConflictError(
                "A referenced reusable definition must declare exactly one "
                "inline template input binding."
            )
        typed = graph_to_ir(nested)
        typed_by_id = {item.id: item for item in typed.nodes}
        output_nodes = [item for item in nested.nodes if item.type == "output"]
        if len(output_nodes) != 1:
            raise DataflowConflictError(
                "A referenced reusable definition must have exactly one "
                "typed output."
            )
        input_contract = _typed_contract(
            typed_by_id[input_nodes[0].id].output_schema,
            label="input",
        )
        output_contract = _typed_contract(
            typed_by_id[output_nodes[0].id].output_schema,
            label="output",
        )
        config = {
            **node.config,
            "template_ref": f"pipeline:{source.id}",
            "template_version": str(source_revision.revision),
            "template_hash": source_revision.content_hash,
            "graph": canonical_graph_payload(nested),
            "input_schema": input_contract,
            "output_schema": output_contract,
            "reference_provenance": {
                "source_scope": {
                    "scope_type": source.scope_type,
                    "scope_id": source.scope_id,
                },
                "source_definition_kind": source.definition_kind,
                "policy_decision": reuse_decision.to_dict(),
            },
        }
        resolved_nodes.append(
            node.model_copy(update={"config": config}, deep=True)
        )
    return graph.model_copy(update={"nodes": resolved_nodes}, deep=True)


def _pipeline_id_from_ref(value: object) -> str:
    text = str(value or "").strip()
    if not text.startswith("pipeline:") or len(text) <= len("pipeline:"):
        raise DataflowConflictError(
            "Reusable subflows require a canonical pipeline reference."
        )
    return text.removeprefix("pipeline:")


def _subflow_revision(value: object) -> int:
    try:
        revision = int(str(value).strip())
    except (TypeError, ValueError) as exc:
        raise DataflowConflictError(
            "Reusable subflows require a valid immutable source revision."
        ) from exc
    if revision < 1:
        raise DataflowConflictError(
            "Reusable subflow revisions must be positive."
        )
    return revision


def _typed_contract(schema: object, *, label: str) -> list[dict[str, object]]:
    fields = tuple(getattr(schema, "fields", ()))
    if not fields or any(getattr(item, "type", "unknown") == "unknown" for item in fields):
        raise DataflowConflictError(
            f"The reusable definition needs a closed typed {label} contract. "
            "Provide representative typed rows at its template input."
        )
    return [
        {
            "name": str(item.name),
            "type": str(item.type),
            "nullable": bool(item.nullable),
        }
        for item in fields
    ]


def create_pipeline(
    session: Session,
    *,
    tenant_id: str,
    actor_id: str | None,
    payload: PipelineCreateRequest,
    principal: ApiPrincipal | None = None,
    registry: object | None = None,
) -> DataflowPipeline:
    pipeline_id = new_uuid()
    graph = _resolve_reusable_subflows(
        session,
        tenant_id=tenant_id,
        graph=payload.graph,
        principal=principal,
        registry=registry,
        target_pipeline_id=pipeline_id,
    )
    definition = normalize_definition(
        graph=graph,
        sql_text=payload.sql_text,
        editor_mode=payload.editor_mode,
    )
    content_hash = definition_hash(definition.graph, definition.sql_text)
    stored_tenant_id = None if payload.scope_type == "system" else tenant_id
    scope_id = (
        None
        if payload.scope_type == "system"
        else tenant_id
        if payload.scope_type == "tenant"
        else payload.scope_id
    )
    pipeline = DataflowPipeline(
        id=pipeline_id,
        tenant_id=stored_tenant_id,
        scope_type=payload.scope_type,
        scope_id=scope_id,
        definition_kind=payload.definition_kind,
        inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
        allow_run=payload.allow_run,
        allow_reuse=payload.allow_reuse,
        allow_automation=payload.allow_automation,
        name=payload.name.strip(),
        description=_clean_optional(payload.description),
        status=payload.status,
        current_revision=1,
        created_by=actor_id,
        updated_by=actor_id,
        metadata_={},
    )
    revision = DataflowPipelineRevision(
        tenant_id=stored_tenant_id,
        revision=1,
        schema_version=definition.graph.schema_version,
        graph=canonical_graph_payload(definition.graph),
        sql_text=definition.sql_text,
        editor_mode=payload.editor_mode,
        content_hash=content_hash,
        created_by=actor_id,
    )
    pipeline.revisions.append(revision)
    session.add(pipeline)
    session.flush()
    return pipeline


def update_pipeline(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
    actor_id: str | None,
    payload: PipelineUpdateRequest,
    principal: ApiPrincipal | None = None,
    registry: object | None = None,
) -> DataflowPipeline:
    pipeline = get_pipeline(session, tenant_id=tenant_id, pipeline_id=pipeline_id)
    if payload.expected_revision != pipeline.current_revision:
        raise DataflowConflictError(
            f"Pipeline changed on the server; expected revision {payload.expected_revision}, "
            f"current revision is {pipeline.current_revision}"
        )
    if (
        payload.scope_type != pipeline.scope_type
        or payload.scope_id != pipeline.scope_id
        and not (
            pipeline.scope_type == "tenant"
            and payload.scope_id in {None, pipeline.scope_id}
        )
    ):
        raise DataflowConflictError(
            "Definition scope is immutable; derive a scoped copy instead."
        )
    if payload.definition_kind != pipeline.definition_kind:
        raise DataflowConflictError(
            "Definition kind is immutable; derive a flow or template instead."
        )
    graph = _resolve_reusable_subflows(
        session,
        tenant_id=tenant_id,
        graph=payload.graph,
        principal=principal,
        registry=registry,
        target_pipeline_id=pipeline.id,
    )
    definition = normalize_definition(
        graph=graph,
        sql_text=payload.sql_text,
        editor_mode=payload.editor_mode,
    )
    content_hash = definition_hash(definition.graph, definition.sql_text)
    current = get_pipeline_revision(session, pipeline=pipeline)
    pipeline.name = payload.name.strip()
    pipeline.description = _clean_optional(payload.description)
    pipeline.status = payload.status
    ancestor_limits = _ancestor_governance_limits(
        pipeline.derivation_provenance
    )
    pipeline.inherit_to_lower_scopes = (
        payload.inherit_to_lower_scopes
        and ancestor_limits["inherit_to_lower_scopes"]
    )
    pipeline.allow_run = payload.allow_run and ancestor_limits["allow_run"]
    pipeline.allow_reuse = (
        payload.allow_reuse and ancestor_limits["allow_reuse"]
    )
    pipeline.allow_automation = (
        payload.allow_automation and ancestor_limits["allow_automation"]
    )
    pipeline.updated_by = actor_id
    if current.content_hash != content_hash or current.editor_mode != payload.editor_mode:
        pipeline.current_revision += 1
        pipeline.revisions.append(
            DataflowPipelineRevision(
                tenant_id=pipeline.tenant_id,
                revision=pipeline.current_revision,
                schema_version=definition.graph.schema_version,
                graph=canonical_graph_payload(definition.graph),
                sql_text=definition.sql_text,
                editor_mode=payload.editor_mode,
                content_hash=content_hash,
                created_by=actor_id,
            )
        )
    session.flush()
    return pipeline


def derive_pipeline(
    session: Session,
    *,
    tenant_id: str,
    actor_id: str | None,
    principal: ApiPrincipal,
    registry: object | None,
    source_pipeline_id: str,
    payload: PipelineDeriveRequest,
) -> DataflowPipeline:
    source = get_pipeline(
        session,
        tenant_id=tenant_id,
        pipeline_id=source_pipeline_id,
    )
    reuse_decision = require_definition_action(
        source,
        principal=principal,
        registry=registry,
        action="derive",
    )
    source_revision = get_pipeline_revision(
        session,
        pipeline=source,
        revision=payload.source_revision,
    )
    stored_tenant_id = None if payload.scope_type == "system" else tenant_id
    scope_id = (
        None
        if payload.scope_type == "system"
        else tenant_id
        if payload.scope_type == "tenant"
        else payload.scope_id
    )
    source_limits = _effective_governance_limits(
        source,
        decision_details=reuse_decision.details,
    )
    effective_limits = {
        "inherit_to_lower_scopes": (
            source_limits["inherit_to_lower_scopes"]
            and payload.inherit_to_lower_scopes
        ),
        "allow_run": source_limits["allow_run"] and payload.allow_run,
        "allow_reuse": source_limits["allow_reuse"] and payload.allow_reuse,
        "allow_automation": (
            source_limits["allow_automation"]
            and payload.allow_automation
        ),
    }
    provenance = {
        "source_ref": f"pipeline:{source.id}",
        "source_scope": {
            "scope_type": source.scope_type,
            "scope_id": source.scope_id,
        },
        "source_definition_kind": source.definition_kind,
        "source_revision": source_revision.revision,
        "source_hash": source_revision.content_hash,
        "source_effective_limits": effective_limits,
        "policy_decision": reuse_decision.to_dict(),
        "derived_by": actor_id,
        "derived_at": utcnow().isoformat(),
    }
    pipeline = DataflowPipeline(
        tenant_id=stored_tenant_id,
        scope_type=payload.scope_type,
        scope_id=scope_id,
        definition_kind=payload.definition_kind,
        inherit_to_lower_scopes=effective_limits[
            "inherit_to_lower_scopes"
        ],
        allow_run=effective_limits["allow_run"],
        allow_reuse=effective_limits["allow_reuse"],
        allow_automation=effective_limits["allow_automation"],
        derived_from_pipeline_id=source.id,
        derived_from_revision=source_revision.revision,
        derived_from_hash=source_revision.content_hash,
        derivation_provenance=provenance,
        name=payload.name.strip(),
        description=_clean_optional(payload.description),
        status="draft",
        current_revision=1,
        created_by=actor_id,
        updated_by=actor_id,
        metadata_={},
    )
    pipeline.revisions.append(
        DataflowPipelineRevision(
            tenant_id=stored_tenant_id,
            revision=1,
            schema_version=source_revision.schema_version,
            graph=dict(source_revision.graph),
            sql_text=source_revision.sql_text,
            editor_mode=source_revision.editor_mode,
            content_hash=source_revision.content_hash,
            created_by=actor_id,
        )
    )
    session.add(pipeline)
    session.flush()
    return pipeline


def pipeline_source_update_status(
    session: Session,
    *,
    tenant_id: str,
    pipeline: DataflowPipeline,
) -> dict[str, object]:
    source_id = pipeline.derived_from_pipeline_id
    if not source_id:
        return {
            "source_available": False,
            "source_name": None,
            "source_current_revision": None,
            "source_current_hash": None,
            "update_available": False,
        }
    source = session.scalar(
        select(DataflowPipeline).where(
            DataflowPipeline.id == source_id,
            or_(
                DataflowPipeline.tenant_id == tenant_id,
                DataflowPipeline.tenant_id.is_(None),
            ),
            DataflowPipeline.deleted_at.is_(None),
        )
    )
    if source is None:
        return {
            "source_available": False,
            "source_name": None,
            "source_current_revision": None,
            "source_current_hash": None,
            "update_available": False,
        }
    revision = get_pipeline_revision(session, pipeline=source)
    return {
        "source_available": True,
        "source_name": source.name,
        "source_current_revision": revision.revision,
        "source_current_hash": revision.content_hash,
        "update_available": (
            revision.revision != pipeline.derived_from_revision
            or revision.content_hash != pipeline.derived_from_hash
        ),
    }


def rebase_pipeline(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
    actor_id: str | None,
    principal: ApiPrincipal,
    registry: object | None,
    payload: PipelineRebaseRequest,
) -> DataflowPipeline:
    pipeline = get_pipeline(
        session,
        tenant_id=tenant_id,
        pipeline_id=pipeline_id,
    )
    if payload.expected_revision != pipeline.current_revision:
        raise DataflowConflictError(
            "Derived pipeline changed on the server; expected revision "
            f"{payload.expected_revision}, current revision is "
            f"{pipeline.current_revision}."
        )
    source_id = pipeline.derived_from_pipeline_id
    if not source_id:
        raise DataflowConflictError(
            "Only a pipeline derived from another definition can adopt a "
            "source update."
        )
    source = get_pipeline(
        session,
        tenant_id=tenant_id,
        pipeline_id=source_id,
    )
    reuse_decision = require_definition_action(
        source,
        principal=principal,
        registry=registry,
        action="derive",
    )
    source_revision = get_pipeline_revision(
        session,
        pipeline=source,
        revision=payload.source_revision,
    )
    if source_revision.content_hash != payload.source_hash:
        raise DataflowConflictError(
            "The reviewed source hash no longer matches the requested "
            "revision; reload before adopting the update."
        )
    if (
        pipeline.derived_from_revision is not None
        and source_revision.revision <= pipeline.derived_from_revision
    ):
        raise DataflowConflictError(
            "A source update must use a revision newer than the currently "
            "pinned revision."
        )

    previous_child_revision = pipeline.current_revision
    previous_child = get_pipeline_revision(session, pipeline=pipeline)
    previous_source_revision = pipeline.derived_from_revision
    previous_source_hash = pipeline.derived_from_hash
    source_limits = _effective_governance_limits(
        source,
        decision_details=reuse_decision.details,
    )
    effective_limits = {
        "inherit_to_lower_scopes": (
            pipeline.inherit_to_lower_scopes
            and source_limits["inherit_to_lower_scopes"]
        ),
        "allow_run": pipeline.allow_run and source_limits["allow_run"],
        "allow_reuse": pipeline.allow_reuse and source_limits["allow_reuse"],
        "allow_automation": (
            pipeline.allow_automation and source_limits["allow_automation"]
        ),
    }
    next_child_revision = previous_child_revision + 1
    rebased_at = utcnow()
    history_value = pipeline.derivation_provenance.get("rebase_history", [])
    history = list(history_value) if isinstance(history_value, list) else []
    history.append(
        {
            "child_revision_before": previous_child_revision,
            "child_hash_before": previous_child.content_hash,
            "child_revision_after": next_child_revision,
            "source_revision_before": previous_source_revision,
            "source_hash_before": previous_source_hash,
            "source_revision_after": source_revision.revision,
            "source_hash_after": source_revision.content_hash,
            "policy_decision": reuse_decision.to_dict(),
            "reason": payload.reason.strip(),
            "rebased_by": actor_id,
            "rebased_at": rebased_at.isoformat(),
        }
    )
    provenance = dict(pipeline.derivation_provenance)
    provenance.update(
        {
            "source_ref": f"pipeline:{source.id}",
            "source_scope": {
                "scope_type": source.scope_type,
                "scope_id": source.scope_id,
            },
            "source_definition_kind": source.definition_kind,
            "source_revision": source_revision.revision,
            "source_hash": source_revision.content_hash,
            "source_effective_limits": effective_limits,
            "policy_decision": reuse_decision.to_dict(),
            "last_rebased_by": actor_id,
            "last_rebased_at": rebased_at.isoformat(),
            "last_rebase_reason": payload.reason.strip(),
            "rebase_history": history,
        }
    )

    pipeline.current_revision = next_child_revision
    pipeline.status = "draft"
    pipeline.inherit_to_lower_scopes = effective_limits[
        "inherit_to_lower_scopes"
    ]
    pipeline.allow_run = effective_limits["allow_run"]
    pipeline.allow_reuse = effective_limits["allow_reuse"]
    pipeline.allow_automation = effective_limits["allow_automation"]
    pipeline.derived_from_revision = source_revision.revision
    pipeline.derived_from_hash = source_revision.content_hash
    pipeline.derivation_provenance = provenance
    pipeline.updated_by = actor_id
    pipeline.revisions.append(
        DataflowPipelineRevision(
            tenant_id=pipeline.tenant_id,
            revision=next_child_revision,
            schema_version=source_revision.schema_version,
            graph=json.loads(json.dumps(source_revision.graph)),
            sql_text=source_revision.sql_text,
            editor_mode=source_revision.editor_mode,
            content_hash=source_revision.content_hash,
            created_by=actor_id,
        )
    )
    session.flush()
    return pipeline


def delete_pipeline(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
    actor_id: str | None,
) -> DataflowPipeline:
    pipeline = get_pipeline(session, tenant_id=tenant_id, pipeline_id=pipeline_id)
    pipeline.deleted_at = utcnow()
    pipeline.updated_by = actor_id
    session.flush()
    return pipeline


def pipeline_response(
    session: Session,
    pipeline: DataflowPipeline,
    *,
    principal: ApiPrincipal,
    registry: object | None,
) -> PipelineResponse:
    revision = get_pipeline_revision(session, pipeline=pipeline)
    return PipelineResponse(
        id=pipeline.id,
        tenant_id=pipeline.tenant_id,
        name=pipeline.name,
        description=pipeline.description,
        status=pipeline.status,
        current_revision=pipeline.current_revision,
        created_by=pipeline.created_by,
        updated_by=pipeline.updated_by,
        created_at=pipeline.created_at,
        updated_at=pipeline.updated_at,
        revision=PipelineRevisionResponse.model_validate(revision),
        governance=definition_governance_payload(
            pipeline,
            principal=principal,
            registry=registry,
            source_update=pipeline_source_update_status(
                session,
                tenant_id=principal.tenant_id,
                pipeline=pipeline,
            ),
        ),
    )


def validate_draft(
    payload: PipelineDraftRequest,
    *,
    session: Session | None = None,
    tenant_id: str | None = None,
    principal: ApiPrincipal | None = None,
    registry: object | None = None,
) -> PipelineValidationResponse:
    if payload.graph is not None and session is not None and tenant_id is not None:
        payload = payload.model_copy(
            update={
                "graph": _resolve_reusable_subflows(
                    session,
                    tenant_id=tenant_id,
                    graph=payload.graph,
                    principal=principal,
                    registry=registry,
                    target_pipeline_id=payload.pipeline_id,
                )
            }
        )
    if payload.sql_text and payload.sql_text.strip():
        try:
            graph, sql_text, diagnostics = compile_sql(
                payload.sql_text,
                source_nodes=_source_nodes(payload.graph, payload.source_nodes),
            )
        except SqlCompilationError as exc:
            return PipelineValidationResponse(
                valid=False,
                graph=payload.graph,
                sql_text=payload.sql_text,
                diagnostics=exc.diagnostics,
            )
        return PipelineValidationResponse(
            valid=True,
            graph=graph,
            sql_text=sql_text,
            diagnostics=diagnostics,
        )
    if payload.graph is None:
        diagnostic = DataflowDiagnostic(
            severity="error",
            code="definition.required",
            message="Provide a graph or SQL query.",
        )
        return PipelineValidationResponse(
            valid=False,
            graph=None,
            sql_text=None,
            diagnostics=[diagnostic],
        )
    diagnostics = validate_graph(payload.graph)
    sql_text: str | None = None
    if not any(item.severity == "error" for item in diagnostics):
        try:
            sql_text, render_diagnostics = render_sql(payload.graph)
            diagnostics.extend(render_diagnostics)
        except SqlCompilationError as exc:
            diagnostics.extend(
                DataflowDiagnostic(
                    severity="warning",
                    code=item.code,
                    message=item.message,
                    node_id=item.node_id,
                    field=item.field,
                )
                for item in exc.diagnostics
            )
    return PipelineValidationResponse(
        valid=not any(item.severity == "error" for item in diagnostics),
        graph=payload.graph,
        sql_text=sql_text,
        diagnostics=diagnostics,
    )


def compile_sql_draft(payload: PipelineDraftRequest) -> PipelineSqlResponse:
    if not payload.sql_text:
        diagnostic = DataflowDiagnostic(
            severity="error",
            code="sql.empty",
            message="Enter a SELECT query.",
            field="sql_text",
        )
        return PipelineSqlResponse(valid=False, graph=payload.graph, sql_text="", diagnostics=[diagnostic])
    try:
        graph, sql_text, diagnostics = compile_sql(
            payload.sql_text,
            source_nodes=_source_nodes(payload.graph, payload.source_nodes),
        )
        if payload.graph is not None:
            graph = preserve_compatible_graph_layout(payload.graph, graph)
    except SqlCompilationError as exc:
        return PipelineSqlResponse(
            valid=False,
            graph=payload.graph,
            sql_text=payload.sql_text,
            diagnostics=exc.diagnostics,
        )
    return PipelineSqlResponse(valid=True, graph=graph, sql_text=sql_text, diagnostics=diagnostics)


def render_graph_sql(payload: PipelineDraftRequest) -> PipelineSqlResponse:
    if payload.graph is None:
        diagnostic = DataflowDiagnostic(
            severity="error",
            code="graph.required",
            message="Provide a graph to render.",
        )
        return PipelineSqlResponse(valid=False, graph=None, sql_text=None, diagnostics=[diagnostic])
    try:
        sql_text, diagnostics = render_sql(payload.graph)
    except SqlCompilationError as exc:
        return PipelineSqlResponse(
            valid=False,
            graph=payload.graph,
            sql_text=None,
            diagnostics=exc.diagnostics,
        )
    return PipelineSqlResponse(valid=True, graph=payload.graph, sql_text=sql_text, diagnostics=diagnostics)


def preview_pipeline(
    session: Session,
    *,
    tenant_id: str,
    actor_id: str | None,
    payload: PipelinePreviewRequest,
    principal: ApiPrincipal | None = None,
    registry: object | None = None,
) -> PipelinePreviewResponse:
    pipeline: DataflowPipeline | None = None
    revision: DataflowPipelineRevision | None = None
    if payload.pipeline_id:
        pipeline = get_pipeline(session, tenant_id=tenant_id, pipeline_id=payload.pipeline_id)
        if principal is None:
            raise DataflowConflictError(
                "Saved pipeline previews require a tenant API principal."
            )
        action = "edit" if pipeline.status == "draft" else "run"
        try:
            require_definition_action(
                pipeline,
                principal=principal,
                registry=registry,
                action=action,
            )
        except PermissionError as exc:
            raise DataflowConflictError(str(exc)) from exc
        revision = get_pipeline_revision(session, pipeline=pipeline, revision=payload.revision)
        graph = PipelineGraph.model_validate(revision.graph)
        sql_text = revision.sql_text
    else:
        draft = PipelineDraftRequest(
            graph=payload.graph,
            sql_text=payload.sql_text,
            source_nodes=payload.source_nodes,
        )
        validated = validate_draft(
            draft,
            session=session,
            tenant_id=tenant_id,
            principal=principal,
            registry=registry,
        )
        if not validated.valid or validated.graph is None:
            return PipelinePreviewResponse(
                run_id=None,
                pipeline_id=None,
                revision=None,
                status="failed",
                columns=[],
                rows=[],
                total_rows=0,
                truncated=False,
                diagnostics=validated.diagnostics,
                node_diagnostics=[],
                node_preview=None,
                source_fingerprints=[],
                input_row_count=0,
                definition_hash="",
                executor_version=EXECUTOR_VERSION,
            )
        graph = validated.graph
        sql_text = validated.sql_text

    graph_hash = definition_hash(graph, sql_text)
    started_at = utcnow()
    run: DataflowRun | None = None
    try:
        result, executor_version = _execute_pipeline_preview(
            graph,
            session=session,
            principal=principal,
            registry=registry,
            backend=payload.execution_backend,
            row_limit=payload.row_limit,
            preview_node_id=payload.preview_node_id,
        )
        status = "succeeded"
        error = None
        diagnostics = result.diagnostics
        columns = result.columns
        rows = result.rows
        total_rows = result.total_rows
        truncated = result.truncated
        node_diagnostics = result.node_diagnostics
        node_preview = result.node_preview
        source_fingerprints = result.source_fingerprints
        input_row_count = result.input_row_count
    except PipelineExecutionError as exc:
        executor_version = (
            payload.execution_backend
            if payload.execution_backend != "reference"
            else EXECUTOR_VERSION
        )
        status = "failed"
        error = str(exc)
        diagnostics = [
            *exc.diagnostics,
            DataflowDiagnostic(
                severity="error",
                code="preview.execution",
                message=str(exc),
                node_id=exc.node_id,
            )
        ]
        columns = []
        rows = []
        total_rows = 0
        truncated = False
        node_diagnostics = list(exc.node_diagnostics)
        node_preview = exc.node_preview
        source_fingerprints = list(exc.source_fingerprints)
        input_row_count = exc.input_row_count

    if pipeline is not None and revision is not None:
        run = DataflowRun(
            tenant_id=tenant_id,
            pipeline_id=pipeline.id,
            pipeline_revision_id=revision.id,
            run_type="preview",
            status=status,
            executor_version=executor_version,
            definition_hash=graph_hash,
            source_fingerprints=source_fingerprints,
            result_schema=[item.model_dump(mode="json") for item in columns],
            diagnostics=[item.model_dump(mode="json") for item in diagnostics],
            input_row_count=input_row_count,
            output_row_count=total_rows,
            started_at=started_at,
            finished_at=utcnow(),
            error=error,
            created_by=actor_id,
        )
        session.add(run)
        session.flush()

    return PipelinePreviewResponse(
        run_id=run.id if run else None,
        pipeline_id=pipeline.id if pipeline else None,
        revision=revision.revision if revision else None,
        status=status,
        columns=columns,
        rows=rows,
        total_rows=total_rows,
        truncated=truncated,
        diagnostics=diagnostics,
        node_diagnostics=node_diagnostics,
        node_preview=node_preview,
        source_fingerprints=source_fingerprints,
        input_row_count=input_row_count,
        definition_hash=graph_hash,
        executor_version=executor_version,
    )


def _execute_pipeline_preview(
    graph: PipelineGraph,
    *,
    session: Session,
    principal: ApiPrincipal | None,
    registry: object | None,
    backend: str,
    row_limit: int,
    preview_node_id: str | None,
    budget: ExecutionBudget | None = None,
) -> tuple[PipelineExecutionResult, str]:
    source_resolver = _preview_source_resolver(
        session=session,
        principal=principal,
        registry=registry,
    )
    if backend == "reference":
        return (
            execute_preview(
                graph,
                row_limit=row_limit,
                source_resolver=source_resolver,
                preview_node_id=preview_node_id,
            ),
            EXECUTOR_VERSION,
        )
    sources = _typed_backend_sources(
        graph,
        source_resolver=source_resolver,
        source_limit=max(MAX_SOURCE_ROWS, row_limit),
    )
    try:
        result = execute_typed_graph(
            graph,
            backend=backend,
            sources=sources,
            budget=budget or ExecutionBudget(max_output_rows=row_limit),
            preview_node_id=preview_node_id,
        )
    except BackendExecutionError as exc:
        raise PipelineExecutionError(
            str(exc),
            node_id=exc.node_id,
            diagnostics=tuple(exc.diagnostics),
            retryable=exc.code == "backend.capacity",
        ) from exc
    columns = [
        PreviewColumn(
            name=field.name,
            type=field.type,
            nullable=field.nullable,
        )
        for field in result.batch.schema.fields
    ]
    source_fingerprints = list(
        result.metadata.get(
            "source_fingerprints",
            result.contract.lineage.source_fingerprints,
        )
    )
    node_preview = _typed_node_preview(
        result,
        preview_node_id=preview_node_id,
        columns=columns,
    )
    return (
        PipelineExecutionResult(
            rows=result.rows,
            total_rows=result.contract.row_count,
            truncated=result.contract.truncated,
            columns=columns,
            diagnostics=list(result.contract.diagnostics),
            node_diagnostics=list(result.node_diagnostics),
            node_preview=node_preview,
            source_fingerprints=source_fingerprints,
            input_row_count=int(
                result.metadata.get(
                    "input_row_count",
                    sum(
                        int(
                            item.get(
                                "row_count",
                                item.get("total_rows", 0),
                            )
                        )
                        for item in source_fingerprints
                    ),
                )
            ),
        ),
        result.contract.backend_version,
    )


def _preview_source_resolver(
    *,
    session: Session,
    principal: ApiPrincipal | None,
    registry: object | None,
):
    if principal is not None:
        return _datasource_source_resolver(
            session=session,
            principal=principal,
            registry=registry,
        )

    def unavailable(node: GraphNode, _limit: int) -> ResolvedSource:
        raise PipelineExecutionError(
            "Datasource-backed preview requires a tenant API principal.",
            node_id=node.id,
        )

    return unavailable


def _typed_backend_sources(
    graph: PipelineGraph,
    *,
    source_resolver,
    source_limit: int = MAX_SOURCE_ROWS,
) -> dict[str, BackendSource]:
    sources: dict[str, BackendSource] = {}
    for node in graph.nodes:
        if node.type != "source.reference":
            continue
        resolved = source_resolver(node, source_limit)
        sources[node.id] = BackendSource(
            node_id=node.id,
            batch=TypedBatch.from_rows(resolved.rows),
            source_ref=resolved.source_ref,
            provider=resolved.provider,
            fingerprint=resolved.fingerprint,
            total_rows=resolved.total_rows,
            truncated=resolved.truncated,
            source_name=str(node.config.get("source_name") or ""),
        )
    return sources


def _typed_node_preview(
    result,
    *,
    preview_node_id: str | None,
    columns: list[PreviewColumn],
) -> NodePreviewResult | None:
    if result.node_preview is not None:
        return result.node_preview
    if preview_node_id is None:
        return None
    return NodePreviewResult(
        node_id=preview_node_id,
        columns=columns,
        rows=result.rows,
        total_rows=result.contract.row_count,
        truncated=result.contract.truncated,
    )


def list_pipeline_runs(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str | None = None,
    limit: int = 100,
) -> list[DataflowRun]:
    statement = (
        select(DataflowRun)
        .where(DataflowRun.tenant_id == tenant_id)
        .order_by(DataflowRun.created_at.desc(), DataflowRun.id.desc())
        .limit(max(1, min(int(limit), 100)))
    )
    if pipeline_id:
        get_pipeline(
            session,
            tenant_id=tenant_id,
            pipeline_id=pipeline_id,
        )
        statement = statement.where(DataflowRun.pipeline_id == pipeline_id)
    return list(session.scalars(statement))


def list_pipeline_deployments(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
) -> list[DataflowPipelineDeployment]:
    get_pipeline(
        session,
        tenant_id=tenant_id,
        pipeline_id=pipeline_id,
    )
    return list(
        session.scalars(
            select(DataflowPipelineDeployment)
            .where(
                DataflowPipelineDeployment.tenant_id == tenant_id,
                DataflowPipelineDeployment.pipeline_id == pipeline_id,
                DataflowPipelineDeployment.status == "active",
            )
            .order_by(DataflowPipelineDeployment.environment)
        )
    )


def promote_pipeline(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
    actor_id: str | None,
    principal: ApiPrincipal,
    registry: object | None,
    payload: PipelinePromotionRequest,
) -> DataflowPipelineDeployment:
    pipeline = get_pipeline(
        session,
        tenant_id=tenant_id,
        pipeline_id=pipeline_id,
    )
    try:
        require_definition_action(
            pipeline,
            principal=principal,
            registry=registry,
            action="edit",
        )
    except PermissionError as exc:
        raise DataflowConflictError(str(exc)) from exc
    revision = get_pipeline_revision(
        session,
        pipeline=pipeline,
        revision=payload.revision,
    )
    if payload.source_environment == "staging":
        source = session.scalar(
            select(DataflowPipelineDeployment).where(
                DataflowPipelineDeployment.tenant_id == tenant_id,
                DataflowPipelineDeployment.pipeline_id == pipeline.id,
                DataflowPipelineDeployment.environment == "staging",
                DataflowPipelineDeployment.status == "active",
            )
        )
        if source is None or source.pipeline_revision_id != revision.id:
            raise DataflowConflictError(
                "Only the revision currently promoted to staging can be "
                "promoted to production."
            )
    deployment = session.scalar(
        select(DataflowPipelineDeployment).where(
            DataflowPipelineDeployment.tenant_id == tenant_id,
            DataflowPipelineDeployment.pipeline_id == pipeline.id,
            DataflowPipelineDeployment.environment
            == payload.target_environment,
        )
    )
    previous_revision_id = (
        deployment.pipeline_revision_id if deployment is not None else None
    )
    if deployment is None:
        deployment = DataflowPipelineDeployment(
            tenant_id=tenant_id,
            pipeline_id=pipeline.id,
            pipeline_revision_id=revision.id,
            environment=payload.target_environment,
            source_environment=payload.source_environment,
            status="active",
            promoted_by=actor_id,
        )
        session.add(deployment)
    deployment.pipeline_revision_id = revision.id
    deployment.source_environment = payload.source_environment
    deployment.status = "active"
    deployment.promoted_by = actor_id
    deployment.provenance = {
        "pipeline_ref": f"pipeline:{pipeline.id}",
        "revision": revision.revision,
        "definition_hash": revision.content_hash,
        "source_environment": payload.source_environment,
        "target_environment": payload.target_environment,
        "previous_revision_id": previous_revision_id,
        "promoted_by": actor_id,
        "promoted_at": utcnow().isoformat(),
    }
    session.flush()
    return deployment


def pipeline_deployment_response(
    session: Session,
    deployment: DataflowPipelineDeployment,
) -> PipelineDeploymentResponse:
    revision = session.get(
        DataflowPipelineRevision,
        deployment.pipeline_revision_id,
    )
    if revision is None:
        raise DataflowNotFoundError("Dataflow pipeline revision not found")
    return PipelineDeploymentResponse(
        id=deployment.id,
        pipeline_id=deployment.pipeline_id,
        revision=revision.revision,
        environment=deployment.environment,  # type: ignore[arg-type]
        source_environment=deployment.source_environment,  # type: ignore[arg-type]
        status=deployment.status,
        provenance=dict(deployment.provenance),
        promoted_by=deployment.promoted_by,
        created_at=deployment.created_at,
        updated_at=deployment.updated_at,
    )


def get_pipeline_run(
    session: Session,
    *,
    tenant_id: str,
    run_ref: str,
) -> DataflowRun:
    run_id = _strip_ref(run_ref, "dataflow-run:")
    if not run_id:
        raise DataflowNotFoundError("Dataflow run not found")
    run = session.scalar(
        select(DataflowRun).where(
            DataflowRun.id == run_id,
            DataflowRun.tenant_id == tenant_id,
        )
    )
    if run is None:
        raise DataflowNotFoundError("Dataflow run not found")
    return run


def start_pipeline_run(
    session: Session,
    *,
    tenant_id: str,
    actor_id: str | None,
    principal: ApiPrincipal,
    registry: object | None,
    request: DataflowRunRequest,
    defer_execution: bool = False,
) -> tuple[DataflowRun, bool]:
    pipeline, revision = _run_definition(
        session,
        tenant_id=tenant_id,
        principal=principal,
        registry=registry,
        request=request,
    )
    idempotency_key, request_hash = _validated_run_identity(request)
    existing = _existing_pipeline_run(
        session,
        tenant_id=tenant_id,
        pipeline_id=pipeline.id,
        idempotency_key=idempotency_key,
        request_hash=request_hash,
    )
    if existing is not None:
        return existing, True
    if defer_execution:
        _require_run_queue_capacity(session, tenant_id=tenant_id)
    run = _new_pipeline_run(
        tenant_id=tenant_id,
        actor_id=actor_id,
        pipeline=pipeline,
        revision=revision,
        request=request,
        idempotency_key=idempotency_key,
        request_hash=request_hash,
        principal=principal,
        defer_execution=defer_execution,
    )
    run.authorization_ = {
        **dict(run.authorization_),
        "authorization_ref": (
            request.invocation.trigger_ref or f"dataflow-run:{run.id}"
        ),
    }
    recovery: DataflowRunRecovery | None = None
    if not defer_execution:
        try:
            recovery = begin_dataflow_run_recovery(
                session,
                run=run,
                lease_ttl_seconds=int(
                    float(run.resource_budget.get("max_wall_seconds") or 30.0)
                )
                + 60,
            )
        except DataflowRecoveryError as exc:
            raise DataflowConflictError(str(exc)) from exc
    session.add(run)
    session.flush()
    if defer_execution:
        session.flush()
        return run, False
    _execute_pipeline_run(
        session,
        run=run,
        pipeline=pipeline,
        revision=revision,
        request=request,
        principal=principal,
        registry=registry,
        recovery=recovery,
    )
    if recovery is not None:
        try:
            recovery.finish(session, run=run)
        except DataflowRecoveryError as exc:
            session.rollback()
            if recovery.publication_started:
                persisted = session.get(DataflowRun, run.id)
                if persisted is not None:
                    persisted.status = "outcome_unknown"
                    persisted.finished_at = utcnow()
                    persisted.progress_phase = "outcome_unknown"
                    persisted.error = (
                        "Output publication completed without verifiable "
                        "terminal recovery evidence; reconcile the sink."
                    )
                    session.commit()
            raise DataflowConflictError(str(exc)) from exc
    else:
        session.flush()
    return run, False


def _run_definition(
    session: Session,
    *,
    tenant_id: str,
    principal: ApiPrincipal,
    registry: object | None,
    request: DataflowRunRequest,
) -> tuple[DataflowPipeline, DataflowPipelineRevision]:
    pipeline_id = _strip_ref(request.pipeline_ref, "pipeline:")
    if not pipeline_id:
        raise DataflowNotFoundError("Dataflow pipeline not found")
    pipeline = get_pipeline(
        session,
        tenant_id=tenant_id,
        pipeline_id=pipeline_id,
    )
    action = (
        "run"
        if request.invocation.kind in {"manual", "api", "backfill"}
        else "automate"
    )
    try:
        require_definition_action(
            pipeline,
            principal=principal,
            registry=registry,
            action=action,
        )
    except PermissionError as exc:
        raise DataflowConflictError(str(exc)) from exc
    revision = get_pipeline_revision(
        session,
        pipeline=pipeline,
        revision=request.revision,
    )
    _require_deployed_revision(
        session,
        tenant_id=tenant_id,
        pipeline=pipeline,
        revision=revision,
        environment=request.environment,
    )
    return pipeline, revision


def _validated_run_identity(request: DataflowRunRequest) -> tuple[str, str]:
    idempotency_key = request.idempotency_key.strip()
    if not idempotency_key or len(idempotency_key) > 255:
        raise DataflowConflictError(
            "A Dataflow run idempotency key of at most 255 characters is required."
        )
    if request.row_limit < 1 or request.row_limit > MAX_PRODUCTION_ROWS:
        raise DataflowConflictError(
            "The bounded Dataflow runner supports between 1 and 10,000 "
            "output rows."
        )
    if request.execution_backend not in {"auto", "reference", "duckdb"}:
        raise DataflowConflictError("Unknown Dataflow execution backend.")
    if request.environment not in {"development", "staging", "production"}:
        raise DataflowConflictError("Unknown Dataflow environment.")
    if request.max_attempts < 1 or request.max_attempts > 5:
        raise DataflowConflictError(
            "Dataflow runs support between one and five attempts."
        )
    if request.retention_days < 1 or request.retention_days > 365:
        raise DataflowConflictError(
            "Dataflow run retention must be between one and 365 days."
        )
    return idempotency_key, _run_request_hash(request)


def _require_run_queue_capacity(
    session: Session,
    *,
    tenant_id: str,
) -> None:
    pending = int(
        session.scalar(
            select(func.count())
            .select_from(DataflowRun)
            .where(
                DataflowRun.tenant_id == tenant_id,
                DataflowRun.status.in_(("queued", "retrying", "running")),
            )
        )
        or 0
    )
    if pending >= MAX_PENDING_RUNS_PER_TENANT:
        raise DataflowConflictError(
            "The tenant Dataflow queue is full; wait for an active run to "
            "finish before submitting more work."
        )


def _require_deployed_revision(
    session: Session,
    *,
    tenant_id: str,
    pipeline: DataflowPipeline,
    revision: DataflowPipelineRevision,
    environment: str,
) -> None:
    if environment == "development":
        return
    deployment = session.scalar(
        select(DataflowPipelineDeployment).where(
            DataflowPipelineDeployment.tenant_id == tenant_id,
            DataflowPipelineDeployment.pipeline_id == pipeline.id,
            DataflowPipelineDeployment.environment == environment,
            DataflowPipelineDeployment.status == "active",
        )
    )
    if (
        deployment is None
        or deployment.pipeline_revision_id != revision.id
    ):
        raise DataflowConflictError(
            f"Pipeline revision {revision.revision} is not promoted to "
            f"{environment}."
        )


def _existing_pipeline_run(
    session: Session,
    *,
    tenant_id: str,
    pipeline_id: str,
    idempotency_key: str,
    request_hash: str,
) -> DataflowRun | None:
    existing = session.scalar(
        select(DataflowRun).where(
            DataflowRun.tenant_id == tenant_id,
            DataflowRun.pipeline_id == pipeline_id,
            DataflowRun.idempotency_key == idempotency_key,
        )
    )
    if existing is not None:
        if existing.request_hash != request_hash:
            raise DataflowConflictError(
                "The Dataflow run idempotency key was already used with "
                "different parameters."
            )
    return existing


def _new_pipeline_run(
    *,
    tenant_id: str,
    actor_id: str | None,
    pipeline: DataflowPipeline,
    revision: DataflowPipelineRevision,
    request: DataflowRunRequest,
    idempotency_key: str,
    request_hash: str,
    principal: ApiPrincipal,
    defer_execution: bool,
) -> DataflowRun:
    now = utcnow()
    budget = _run_resource_budget(request)
    return DataflowRun(
        id=new_uuid(),
        tenant_id=tenant_id,
        pipeline_id=pipeline.id,
        pipeline_revision_id=revision.id,
        run_type="published" if request.publication else "run",
        status="queued" if defer_execution else "running",
        execution_backend=request.execution_backend,
        environment=request.environment,
        executor_version=EXECUTOR_VERSION,
        definition_hash=revision.content_hash,
        idempotency_key=idempotency_key,
        request_hash=request_hash,
        request_=_run_request_payload(request),
        invocation_kind=request.invocation.kind,
        trigger_id=_strip_ref(
            request.invocation.trigger_ref or "",
            "dataflow-trigger:",
        ),
        trigger_delivery_id=_strip_ref(
            request.invocation.delivery_ref or "",
            "dataflow-trigger-delivery:",
        ),
        correlation_id=request.invocation.correlation_id,
        causation_id=request.invocation.causation_id,
        source_fingerprints=[],
        result_schema=[],
        diagnostics=[],
        input_row_count=0,
        output_row_count=0,
        attempts=0 if defer_execution else 1,
        max_attempts=request.max_attempts,
        queued_at=now if defer_execution else None,
        available_at=now if defer_execution else None,
        progress_percent=0 if defer_execution else 10,
        progress_phase="queued" if defer_execution else "executing",
        retention_until=now + timedelta(days=request.retention_days),
        authorization_=_run_authorization_payload(
            principal,
            request=request,
            revision=revision,
        ),
        resource_budget=budget,
        started_at=None if defer_execution else now,
        created_by=actor_id,
    )


def _run_authorization_payload(
    principal: ApiPrincipal,
    *,
    request: DataflowRunRequest,
    revision: DataflowPipelineRevision,
) -> dict[str, object]:
    principal_ref = principal.to_platform_principal()
    graph = PipelineGraph.model_validate(revision.graph)
    scopes = {RUN_SCOPE}
    if any(
        (
            node.type == "source.reference"
            or bool(node.config.get("source_ref"))
        )
        and decision_set_id_from_ref(str(node.config.get("source_ref") or ""))
        is None
        for node in graph.nodes
    ):
        scopes.add(DATASOURCE_READ_SCOPE)
    if request.publication is not None:
        scopes.add(DATASOURCE_WRITE_SCOPE)
    return {
        "contract_version": "1",
        "subject_kind": (
            "service_account"
            if principal_ref.service_account_id
            else "delegated_user"
        ),
        "account_id": principal_ref.account_id,
        "membership_id": principal_ref.membership_id,
        "service_account_id": principal_ref.service_account_id,
        "grant_scopes": sorted(scopes),
        "submitted_principal": principal_ref.to_dict(),
        "authorization_ref": request.invocation.trigger_ref,
        "resolution": "rechecked_by_worker",
    }


def _run_resource_budget(
    request: DataflowRunRequest,
) -> dict[str, object]:
    production = request.environment in {"staging", "production"}
    return {
        "max_output_rows": request.row_limit,
        "max_batch_bytes": 64 * 1024 * 1024 if production else 8 * 1024 * 1024,
        "max_wall_seconds": 30.0 if production else 10.0,
        "max_memory_bytes": (
            512 * 1024 * 1024 if production else 256 * 1024 * 1024
        ),
        "max_concurrency": 1,
    }


def _execute_pipeline_run(
    session: Session,
    *,
    run: DataflowRun,
    pipeline: DataflowPipeline,
    revision: DataflowPipelineRevision,
    request: DataflowRunRequest,
    principal: ApiPrincipal,
    registry: object | None,
    recovery: DataflowRunRecovery | None = None,
) -> bool:
    try:
        if run.cancellation_requested_at is not None:
            _mark_pipeline_run_cancelled(run)
            return False
        execution_backend = run.execution_backend
        if run.environment in {"staging", "production"}:
            if execution_backend == "reference":
                raise PipelineExecutionError(
                    "Staging and production runs require the isolated "
                    "DuckDB execution backend."
                )
            execution_backend = "duckdb"
        run.progress_percent = 20
        run.progress_phase = "reading_sources"
        result, executor_version = _execute_pipeline_preview(
            PipelineGraph.model_validate(revision.graph),
            session=session,
            principal=principal,
            registry=registry,
            backend=execution_backend,
            row_limit=request.row_limit,
            preview_node_id=None,
            budget=_execution_budget(run),
        )
        run.executor_version = executor_version
        _apply_pipeline_result(run, result)
        run.progress_percent = 80
        run.progress_phase = "publishing" if request.publication else "finalizing"
        session.flush()
        session.expire(run, ["cancellation_requested_at"])
        if run.cancellation_requested_at is not None:
            _mark_pipeline_run_cancelled(run)
            return False
        if request.publication:
            _ensure_publishable(result)
            _publish_pipeline_result(
                session,
                run=run,
                pipeline=pipeline,
                revision=revision,
                request=request,
                result=result,
                principal=principal,
                registry=registry,
                recovery=recovery,
            )
        run.status = "succeeded"
        run.finished_at = utcnow()
        run.error = None
        run.progress_percent = 100
        run.progress_phase = "completed"
        return False
    except (DatasourceError, PipelineExecutionError) as exc:
        _mark_pipeline_run_failed(run, exc)
        if recovery is not None and recovery.publication_started:
            run.status = "outcome_unknown"
            run.progress_phase = "outcome_unknown"
            return False
        return isinstance(exc, DatasourceUnavailableError) or bool(
            getattr(exc, "retryable", False)
        )


def _execution_budget(run: DataflowRun) -> ExecutionBudget:
    value = dict(run.resource_budget)
    return ExecutionBudget(
        max_output_rows=int(value.get("max_output_rows") or 500),
        max_batch_bytes=int(value.get("max_batch_bytes") or 8 * 1024 * 1024),
        max_wall_seconds=float(value.get("max_wall_seconds") or 10.0),
        max_memory_bytes=int(
            value.get("max_memory_bytes") or 256 * 1024 * 1024
        ),
        max_concurrency=int(value.get("max_concurrency") or 1),
    )


def _apply_pipeline_result(
    run: DataflowRun,
    result: PipelineExecutionResult,
) -> None:
    run.source_fingerprints = result.source_fingerprints
    run.result_schema = [
        item.model_dump(mode="json") for item in result.columns
    ]
    run.diagnostics = [
        item.model_dump(mode="json") for item in result.diagnostics
    ]
    run.input_row_count = result.input_row_count
    run.output_row_count = result.total_rows


def _ensure_publishable(result: PipelineExecutionResult) -> None:
    source_truncated = any(
        bool(item.get("truncated"))
        for item in result.source_fingerprints
    )
    if result.truncated or source_truncated:
        raise PipelineExecutionError(
            "The bounded runner cannot publish a truncated result or a "
            "result calculated from truncated source data."
        )


def _publish_pipeline_result(
    session: Session,
    *,
    run: DataflowRun,
    pipeline: DataflowPipeline,
    revision: DataflowPipelineRevision,
    request: DataflowRunRequest,
    result: PipelineExecutionResult,
    principal: ApiPrincipal,
    registry: object | None,
    recovery: DataflowRunRecovery | None = None,
) -> None:
    publisher = datasource_publication(registry)
    if publisher is None:
        raise PipelineExecutionError(
            "Publishing Dataflow output requires the Datasources "
            "publication capability."
        )
    target = request.publication
    if target is None:
        return
    if recovery is None:
        raise PipelineExecutionError(
            "Publishing Dataflow output requires a durable recovery operation."
        )
    recovery.prepare_publication(
        session,
        run=run,
        rows=tuple(dict(row) for row in result.rows),
    )
    try:
        publication = publisher.publish_rows(
            session,
            principal,
            request=DatasourcePublicationRequest(
                producer_module="dataflow",
                producer_run_ref=f"dataflow-run:{run.id}",
                idempotency_key=(
                    f"{pipeline.id}:{request.idempotency_key.strip()}"
                ),
                rows=tuple(dict(row) for row in result.rows),
                target_datasource_ref=target.target_datasource_ref,
                name=target.name or f"{pipeline.name} output",
                source_name=target.source_name,
                description=target.description,
                freeze=target.freeze,
                frozen_label=target.frozen_label,
                set_current=target.set_current,
                provenance={
                    "pipeline_ref": f"pipeline:{pipeline.id}",
                    "pipeline_revision": revision.revision,
                    "definition_hash": revision.content_hash,
                    "source_fingerprints": result.source_fingerprints,
                },
                metadata={
                    **dict(target.metadata),
                    "dataflow_run_ref": f"dataflow-run:{run.id}",
                },
            ),
        )
    except DatasourceError:
        raise
    except Exception as exc:
        raise PipelineExecutionError(
            "The output provider failed after publication dispatch began."
        ) from exc
    run.output_publication_ref = publication.ref
    run.output_datasource_ref = publication.datasource.ref
    run.output_materialization_ref = publication.materialization.ref
    if publication.status in {"published_with_warnings", "review_required"}:
        diagnostics = list(run.diagnostics)
        diagnostics.append(
            DataflowDiagnostic(
                severity="warning",
                code=(
                    "publication.review_required"
                    if publication.status == "review_required"
                    else "publication.warning"
                ),
                message=(
                    "The output materialization requires review before it can "
                    "become the datasource's current state."
                    if publication.status == "review_required"
                    else "The output was published with datasource validation "
                    "warnings."
                ),
            ).model_dump(mode="json")
        )
        run.diagnostics = diagnostics


def _mark_pipeline_run_failed(
    run: DataflowRun,
    exc: DatasourceError | PipelineExecutionError,
) -> None:
    run.status = "failed"
    run.finished_at = utcnow()
    run.error = str(exc)
    run.progress_phase = "failed"
    diagnostics = list(getattr(exc, "diagnostics", ()))
    diagnostics.append(
        DataflowDiagnostic(
            severity="error",
            code="run.execution",
            message=str(exc),
            node_id=getattr(exc, "node_id", None),
        )
    )
    run.diagnostics = [
        item.model_dump(mode="json") for item in diagnostics
    ]
    run.source_fingerprints = list(
        getattr(exc, "source_fingerprints", ())
    )
    run.input_row_count = int(getattr(exc, "input_row_count", 0))


def _mark_pipeline_run_cancelled(run: DataflowRun) -> None:
    run.status = "cancelled"
    run.finished_at = utcnow()
    run.error = "Cancelled by request."
    run.progress_phase = "cancelled"


def cancel_pipeline_run(
    session: Session,
    *,
    tenant_id: str,
    run_ref: str,
) -> DataflowRun:
    run = get_pipeline_run(
        session,
        tenant_id=tenant_id,
        run_ref=run_ref,
    )
    if run.status not in {"queued", "retrying", "running"}:
        raise DataflowConflictError(
            f"Dataflow run is already {run.status} and cannot be cancelled."
        )
    run.cancellation_requested_at = utcnow()
    if run.status in {"queued", "retrying"}:
        _mark_pipeline_run_cancelled(run)
    else:
        run.progress_phase = "cancellation_requested"
    session.flush()
    return run


def pipeline_run_response(
    session: Session,
    run: DataflowRun,
    *,
    replayed: bool = False,
    recovery_state: Mapping[str, object] | None = None,
) -> PipelineRunResponse:
    revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id)
    if revision is None:
        raise DataflowNotFoundError("Dataflow pipeline revision not found")
    recovery = (
        dict(recovery_state) or None
        if recovery_state is not None
        else dataflow_run_recovery_state(session, run_id=run.id)
    )
    return PipelineRunResponse(
        ref=f"dataflow-run:{run.id}",
        pipeline_id=run.pipeline_id,
        revision=revision.revision,
        run_type=run.run_type,
        status=run.status,  # type: ignore[arg-type]
        idempotency_key=run.idempotency_key,
        execution_backend=run.execution_backend,
        environment=run.environment,  # type: ignore[arg-type]
        definition_hash=run.definition_hash,
        executor_version=run.executor_version,
        source_fingerprints=list(run.source_fingerprints),
        result_schema=list(run.result_schema),
        diagnostics=list(run.diagnostics),
        input_row_count=run.input_row_count,
        output_row_count=run.output_row_count,
        output_publication_ref=run.output_publication_ref,
        output_datasource_ref=run.output_datasource_ref,
        output_materialization_ref=run.output_materialization_ref,
        invocation_kind=run.invocation_kind,
        trigger_ref=(
            f"dataflow-trigger:{run.trigger_id}" if run.trigger_id else None
        ),
        delivery_ref=(
            f"dataflow-trigger-delivery:{run.trigger_delivery_id}"
            if run.trigger_delivery_id
            else None
        ),
        correlation_id=run.correlation_id,
        causation_id=run.causation_id,
        attempts=run.attempts,
        max_attempts=run.max_attempts,
        available_at=run.available_at,
        claimed_at=run.claimed_at,
        lease_expires_at=run.lease_expires_at,
        cancellation_requested_at=run.cancellation_requested_at,
        progress_percent=run.progress_percent,
        progress_phase=run.progress_phase,
        retention_until=run.retention_until,
        purged_at=run.purged_at,
        error=run.error,
        started_at=run.started_at,
        finished_at=run.finished_at,
        created_by=run.created_by,
        created_at=run.created_at,
        recovery_operation_id=(
            str(recovery["operation_id"]) if recovery is not None else None
        ),
        recovery_operation_type=(
            str(recovery["operation_type"]) if recovery is not None else None
        ),
        recovery_mode=(
            str(recovery["mode"]) if recovery is not None else None
        ),
        recovery_status=(
            str(recovery["status"]) if recovery is not None else None
        ),
        recovery_requires_attention=(
            bool(recovery["requires_attention"])
            or run.status == "outcome_unknown"
            if recovery is not None
            else run.status == "outcome_unknown"
        ),
        recovery_explanation=(
            str(recovery["explanation"]) if recovery is not None else None
        ),
        replayed=replayed,
    )


def pipeline_run_descriptor(
    session: Session,
    run: DataflowRun,
    *,
    replayed: bool = False,
) -> DataflowRunDescriptor:
    revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id)
    if revision is None:
        raise DataflowNotFoundError("Dataflow pipeline revision not found")
    recovery = dataflow_run_recovery_state(session, run_id=run.id)
    return DataflowRunDescriptor(
        ref=f"dataflow-run:{run.id}",
        pipeline_ref=f"pipeline:{run.pipeline_id}",
        revision=revision.revision,
        status=run.status,
        definition_hash=run.definition_hash,
        executor_version=run.executor_version,
        input_row_count=run.input_row_count,
        output_row_count=run.output_row_count,
        output_publication_ref=run.output_publication_ref,
        output_datasource_ref=run.output_datasource_ref,
        output_materialization_ref=run.output_materialization_ref,
        invocation_kind=run.invocation_kind,
        trigger_ref=(
            f"dataflow-trigger:{run.trigger_id}" if run.trigger_id else None
        ),
        delivery_ref=(
            f"dataflow-trigger-delivery:{run.trigger_delivery_id}"
            if run.trigger_delivery_id
            else None
        ),
        error=run.error,
        started_at=run.started_at,
        finished_at=run.finished_at,
        replayed=replayed,
        metadata={
            "run_type": run.run_type,
            "execution_backend": run.execution_backend,
            "environment": run.environment,
            "attempts": run.attempts,
            "max_attempts": run.max_attempts,
            "progress_percent": run.progress_percent,
            "progress_phase": run.progress_phase,
            "source_fingerprints": list(run.source_fingerprints),
            "diagnostics": list(run.diagnostics),
            "recovery": dict(recovery) if recovery is not None else None,
        },
    )


class SqlDataflowRunLifecycleProvider:
    def __init__(self, *, registry: object | None = None) -> None:
        self._registry = registry

    def start_run(
        self,
        session: object,
        principal: object,
        *,
        request: DataflowRunRequest,
    ) -> DataflowRunDescriptor:
        db, api_principal = _run_context(session, principal)
        run, replayed = start_pipeline_run(
            db,
            tenant_id=api_principal.tenant_id,
            actor_id=_principal_actor_id(api_principal),
            principal=api_principal,
            registry=self._registry,
            request=request,
            defer_execution=True,
        )
        return pipeline_run_descriptor(db, run, replayed=replayed)

    def get_run(
        self,
        session: object,
        principal: object,
        *,
        run_ref: str,
    ) -> DataflowRunDescriptor | None:
        db, api_principal = _run_context(session, principal)
        try:
            run = get_pipeline_run(
                db,
                tenant_id=api_principal.tenant_id,
                run_ref=run_ref,
            )
        except DataflowNotFoundError:
            return None
        return pipeline_run_descriptor(db, run)

    def cancel_run(
        self,
        session: object,
        principal: object,
        *,
        run_ref: str,
    ) -> DataflowRunDescriptor:
        db, api_principal = _run_context(session, principal)
        run = cancel_pipeline_run(
            db,
            tenant_id=api_principal.tenant_id,
            run_ref=run_ref,
        )
        return pipeline_run_descriptor(db, run)


def normalize_definition(
    *,
    graph: PipelineGraph,
    sql_text: str | None,
    editor_mode: str,
) -> NormalizedDefinition:
    if editor_mode == "sql":
        try:
            compiled_graph, normalized_sql, diagnostics = compile_sql(
                sql_text or "",
                source_nodes=_source_nodes(graph, ()),
            )
        except SqlCompilationError as exc:
            raise DataflowValidationError(exc.diagnostics) from exc
        return NormalizedDefinition(
            graph=compiled_graph,
            sql_text=normalized_sql,
            diagnostics=diagnostics,
        )
    diagnostics = validate_graph(graph)
    errors = [item for item in diagnostics if item.severity == "error"]
    if errors:
        raise DataflowValidationError(diagnostics)
    try:
        rendered_sql, render_diagnostics = render_sql(graph)
        diagnostics.extend(render_diagnostics)
    except SqlCompilationError:
        rendered_sql = None
    return NormalizedDefinition(graph=graph, sql_text=rendered_sql, diagnostics=diagnostics)


def _source_nodes(
    graph: PipelineGraph | None,
    explicit_nodes: list[GraphNode] | tuple[()],
) -> list[GraphNode]:
    nodes = list(explicit_nodes)
    if graph is not None:
        known = {node.id for node in nodes}
        nodes.extend(
            node
            for node in graph.nodes
            if node.type.startswith("source.") and node.id not in known
        )
    return nodes


def _datasource_source_resolver(
    *,
    session: Session,
    principal: ApiPrincipal,
    registry: object | None,
):
    provider = datasource_catalogue(registry)

    def resolve_source(node: GraphNode, limit: int) -> ResolvedSource:
        source_ref = str(node.config.get("source_ref") or "")
        decision_set_id = decision_set_id_from_ref(source_ref)
        if decision_set_id is not None:
            try:
                decision_set = get_decision_set(
                    session,
                    tenant_id=principal.tenant_id,
                    decision_set_id=decision_set_id,
                    include_decisions=False,
                )
                require_definition_action(
                    decision_set.pipeline,
                    principal=principal,
                    registry=registry,
                    action="view",
                )
            except (PermissionError, ValueError) as exc:
                raise PipelineExecutionError(
                    str(exc),
                    node_id=node.id,
                ) from exc
            rows, total_rows = current_decision_rows(
                session,
                decision_set_id=decision_set.id,
                limit=limit,
            )
            fingerprint = decision_set_fingerprint(decision_set)
            expected_fingerprint = _clean_optional(
                node.config.get("expected_fingerprint")
            )
            if expected_fingerprint and expected_fingerprint != fingerprint:
                raise PipelineExecutionError(
                    "Reconciliation decisions changed; refresh the decision source before running it.",
                    node_id=node.id,
                )
            return ResolvedSource(
                rows=rows,
                source_ref=source_ref,
                provider="dataflow.reconciliation_decisions",
                fingerprint=fingerprint,
                total_rows=total_rows,
                truncated=total_rows > len(rows),
            )
        if provider is None:
            raise PipelineExecutionError(
                "Datasource-backed execution requires the Datasources "
                "catalogue capability.",
                node_id=node.id,
            )
        rows: list[dict[str, object]] = []
        resolved = None
        offset = 0
        expected_fingerprint = _clean_optional(
            node.config.get("expected_fingerprint")
        )
        try:
            while offset < limit:
                page = provider.read_datasource(
                    session,
                    principal,
                    request=DatasourceReadRequest(
                        datasource_ref=str(node.config["source_ref"]),
                        consistency=str(
                            node.config.get("consistency") or "current"
                        ),  # type: ignore[arg-type]
                        limit=min(500, limit - offset),
                        offset=offset,
                        expected_fingerprint=expected_fingerprint,
                    ),
                )
                if resolved is None:
                    resolved = page
                    expected_fingerprint = page.datasource.fingerprint
                elif (
                    page.datasource.fingerprint
                    != resolved.datasource.fingerprint
                ):
                    raise PipelineExecutionError(
                        "Datasource changed while the run was reading it.",
                        node_id=node.id,
                        retryable=True,
                    )
                rows.extend(dict(row) for row in page.rows)
                offset += len(page.rows)
                if not page.truncated or not page.rows:
                    resolved = page
                    break
                resolved = page
        except DatasourceUnavailableError as exc:
            raise PipelineExecutionError(
                str(exc),
                node_id=node.id,
                retryable=True,
            ) from exc
        except DatasourceError as exc:
            raise PipelineExecutionError(str(exc), node_id=node.id) from exc
        if resolved is None:
            raise PipelineExecutionError(
                "Datasource returned no result.",
                node_id=node.id,
                retryable=True,
            )
        return ResolvedSource(
            rows=tuple(rows),
            source_ref=resolved.datasource.ref,
            provider=resolved.datasource.provider or "datasources",
            fingerprint=resolved.datasource.fingerprint,
            total_rows=resolved.total_rows,
            truncated=len(rows) < resolved.total_rows,
        )

    return resolve_source


def _run_context(
    session: object,
    principal: object,
) -> tuple[Session, ApiPrincipal]:
    if not isinstance(session, Session):
        raise TypeError("Dataflow run providers require a SQLAlchemy session.")
    if not isinstance(principal, ApiPrincipal):
        raise DataflowConflictError("A tenant API principal is required.")
    if not principal.tenant_id:
        raise DataflowConflictError("A tenant API principal is required.")
    return session, principal


def _principal_actor_id(principal: ApiPrincipal) -> str | None:
    return principal.account_id or principal.membership_id or principal.identity_id


def _strip_ref(value: str, prefix: str) -> str | None:
    cleaned = str(value or "").strip()
    if not cleaned:
        return None
    if cleaned.startswith(prefix):
        return cleaned[len(prefix) :]
    return cleaned if ":" not in cleaned else None


def _run_request_payload(request: DataflowRunRequest) -> dict[str, object]:
    target = request.publication
    return {
        "pipeline_ref": request.pipeline_ref,
        "revision": request.revision,
        "row_limit": request.row_limit,
        "execution_backend": request.execution_backend,
        "environment": request.environment,
        "max_attempts": request.max_attempts,
        "retention_days": request.retention_days,
        "publication": (
            {
                "target_datasource_ref": target.target_datasource_ref,
                "name": target.name,
                "source_name": target.source_name,
                "description": target.description,
                "freeze": target.freeze,
                "frozen_label": target.frozen_label,
                "set_current": target.set_current,
                "metadata": dict(target.metadata),
            }
            if target
            else None
        ),
        "invocation": _invocation_payload(request.invocation),
    }


def _invocation_payload(
    invocation: AutomationInvocation,
) -> dict[str, object]:
    return {
        "kind": invocation.kind,
        "trigger_ref": invocation.trigger_ref,
        "delivery_ref": invocation.delivery_ref,
        "event_id": invocation.event_id,
        "event_type": invocation.event_type,
        "correlation_id": invocation.correlation_id,
        "causation_id": invocation.causation_id,
        "scheduled_for": (
            invocation.scheduled_for.isoformat()
            if invocation.scheduled_for
            else None
        ),
        "requested_by": invocation.requested_by,
        "metadata": dict(invocation.metadata),
    }


def pipeline_run_request(run: DataflowRun) -> DataflowRunRequest:
    value = dict(run.request_)
    publication_value = value.get("publication")
    publication = (
        DataflowPublicationTarget(
            target_datasource_ref=_mapping_optional_text(
                publication_value,
                "target_datasource_ref",
            ),
            name=_mapping_optional_text(publication_value, "name"),
            source_name=_mapping_optional_text(
                publication_value,
                "source_name",
            ),
            description=_mapping_optional_text(
                publication_value,
                "description",
            ),
            freeze=bool(publication_value.get("freeze", False)),
            frozen_label=_mapping_optional_text(
                publication_value,
                "frozen_label",
            ),
            set_current=bool(publication_value.get("set_current", True)),
            metadata=(
                dict(publication_value.get("metadata") or {})
                if isinstance(publication_value.get("metadata"), Mapping)
                else {}
            ),
        )
        if isinstance(publication_value, Mapping)
        else None
    )
    invocation_value = value.get("invocation")
    invocation_mapping = (
        invocation_value if isinstance(invocation_value, Mapping) else {}
    )
    metadata = invocation_mapping.get("metadata")
    return DataflowRunRequest(
        pipeline_ref=str(value.get("pipeline_ref") or f"pipeline:{run.pipeline_id}"),
        revision=int(value.get("revision") or 1),
        idempotency_key=str(value.get("idempotency_key") or run.idempotency_key or run.id),
        row_limit=int(value.get("row_limit") or 500),
        execution_backend=str(
            value.get("execution_backend") or run.execution_backend
        ),
        environment=str(value.get("environment") or run.environment),
        max_attempts=int(value.get("max_attempts") or run.max_attempts),
        retention_days=int(value.get("retention_days") or 30),
        publication=publication,
        invocation=AutomationInvocation(
            kind=str(
                invocation_mapping.get("kind") or run.invocation_kind
            ),  # type: ignore[arg-type]
            trigger_ref=_mapping_optional_text(
                invocation_mapping,
                "trigger_ref",
            ),
            delivery_ref=_mapping_optional_text(
                invocation_mapping,
                "delivery_ref",
            ),
            event_id=_mapping_optional_text(invocation_mapping, "event_id"),
            event_type=_mapping_optional_text(
                invocation_mapping,
                "event_type",
            ),
            correlation_id=_mapping_optional_text(
                invocation_mapping,
                "correlation_id",
            ),
            causation_id=_mapping_optional_text(
                invocation_mapping,
                "causation_id",
            ),
            scheduled_for=_optional_datetime(
                invocation_mapping.get("scheduled_for")
            ),
            requested_by=_mapping_optional_text(
                invocation_mapping,
                "requested_by",
            ),
            metadata=dict(metadata) if isinstance(metadata, Mapping) else {},
        ),
    )


def _mapping_optional_text(
    value: object,
    key: str,
) -> str | None:
    if not isinstance(value, Mapping):
        return None
    return _clean_optional(value.get(key))


def _optional_datetime(value: object) -> datetime | None:
    if isinstance(value, datetime):
        return value
    if isinstance(value, str) and value.strip():
        try:
            return datetime.fromisoformat(value)
        except ValueError:
            return None
    return None


def _run_request_hash(request: DataflowRunRequest) -> str:
    encoded = json.dumps(
        _run_request_payload(request),
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def _clean_optional(value: object | None) -> str | None:
    if value is None:
        return None
    cleaned = str(value).strip()
    return cleaned or None


def _ancestor_governance_limits(
    provenance: Mapping[str, object],
) -> dict[str, bool]:
    raw = provenance.get("source_effective_limits")
    limits = raw if isinstance(raw, Mapping) else {}
    return {
        key: value if isinstance((value := limits.get(key)), bool) else True
        for key in (
            "inherit_to_lower_scopes",
            "allow_run",
            "allow_reuse",
            "allow_automation",
        )
    }


def _effective_governance_limits(
    pipeline: DataflowPipeline,
    *,
    decision_details: Mapping[str, object] | None = None,
) -> dict[str, bool]:
    ancestor = _ancestor_governance_limits(
        pipeline.derivation_provenance
    )
    effective = {
        "inherit_to_lower_scopes": (
            pipeline.inherit_to_lower_scopes
            and ancestor["inherit_to_lower_scopes"]
        ),
        "allow_run": pipeline.allow_run and ancestor["allow_run"],
        "allow_reuse": pipeline.allow_reuse and ancestor["allow_reuse"],
        "allow_automation": (
            pipeline.allow_automation and ancestor["allow_automation"]
        ),
    }
    policy_limits = (
        decision_details.get("effective_limits")
        if decision_details is not None
        else None
    )
    if isinstance(policy_limits, Mapping):
        for key in effective:
            value = policy_limits.get(key)
            if isinstance(value, bool):
                effective[key] = effective[key] and value
    return effective


__all__ = [
    "DataflowConflictError",
    "DataflowError",
    "DataflowNotFoundError",
    "DataflowValidationError",
    "SqlDataflowRunLifecycleProvider",
    "cancel_pipeline_run",
    "compile_sql_draft",
    "create_pipeline",
    "derive_pipeline",
    "delete_pipeline",
    "get_pipeline",
    "get_pipeline_revision",
    "get_pipeline_run",
    "list_pipeline_deployments",
    "list_pipeline_runs",
    "list_pipelines",
    "normalize_definition",
    "pipeline_response",
    "pipeline_source_update_status",
    "pipeline_deployment_response",
    "pipeline_run_descriptor",
    "pipeline_run_request",
    "pipeline_run_response",
    "promote_pipeline",
    "rebase_pipeline",
    "preview_pipeline",
    "render_graph_sql",
    "start_pipeline_run",
    "update_pipeline",
    "validate_draft",
]