Install and reconcile module workflow standards

src/govoplan_workflow_engine/backend/configuration_provider.py

@@ -0,0 +1,487 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.configuration_packages import (
+    ConfigurationApplyResult,
+    ConfigurationDiagnostic,
+    ConfigurationExportResult,
+    ConfigurationExportSelection,
+    ConfigurationPackageFragment,
+    ConfigurationPlanItem,
+    ConfigurationPreflightContext,
+    ConfigurationPreflightResult,
+    ConfigurationProvider,
+    ConfigurationProviderDescription,
+)
+from govoplan_core.core.workflows import (
+    WorkflowDefinitionContribution,
+    workflow_definition_contribution_hash,
+)
+from govoplan_core.db.session import get_database
+from govoplan_workflow_engine.backend.contributions import (
+    reconcile_workflow_contributions,
+)
+from govoplan_workflow_engine.backend.db.models import (
+    WorkflowDefinition,
+)
+from govoplan_workflow_engine.backend.schemas import WorkflowGraph
+from govoplan_workflow_engine.backend.validation import validate_workflow_graph
+
+
+WORKFLOW_CONFIGURATION_CAPABILITY = "workflow.configuration"
+WORKFLOW_DEFINITIONS_FRAGMENT = "workflow_definitions"
+
+
+class SqlWorkflowConfigurationProvider(ConfigurationProvider):
+    module_id = "workflow_engine"
+
+    def __init__(self, *, registry: object | None = None) -> None:
+        self._registry = registry
+
+    def describe(self) -> ConfigurationProviderDescription:
+        return ConfigurationProviderDescription(
+            module_id=self.module_id,
+            fragment_types=(WORKFLOW_DEFINITIONS_FRAGMENT,),
+            schema_refs={
+                WORKFLOW_DEFINITIONS_FRAGMENT: (
+                    "govoplan/workflow-engine/configuration/definitions.v1"
+                ),
+            },
+            exported_scopes=("system", "tenant"),
+        )
+
+    def preflight(
+        self,
+        fragment: ConfigurationPackageFragment,
+        context: ConfigurationPreflightContext,
+    ) -> ConfigurationPreflightResult:
+        with get_database().session() as session:
+            return preflight_workflow_definitions(session, fragment, context)
+
+    def apply(
+        self,
+        fragment: ConfigurationPackageFragment,
+        supplied_data: Mapping[str, Any],
+        context: ConfigurationPreflightContext,
+    ) -> ConfigurationApplyResult:
+        del supplied_data
+        with get_database().session() as session:
+            result = apply_workflow_definitions(
+                session,
+                fragment,
+                context,
+                registry=self._registry,
+            )
+            if not any(item.severity == "blocker" for item in result.diagnostics):
+                session.commit()
+            return result
+
+    def export(
+        self,
+        selection: ConfigurationExportSelection,
+        context: ConfigurationPreflightContext,
+    ) -> ConfigurationExportResult:
+        del context
+        with get_database().session() as session:
+            return export_workflow_definitions(session, selection)
+
+    def health(
+        self,
+        import_result: ConfigurationApplyResult,
+        context: ConfigurationPreflightContext,
+    ) -> tuple[ConfigurationDiagnostic, ...]:
+        del context
+        return tuple(
+            item for item in import_result.diagnostics if item.severity == "blocker"
+        )
+
+
+def preflight_workflow_definitions(
+    session: Session,
+    fragment: ConfigurationPackageFragment,
+    context: ConfigurationPreflightContext,
+) -> ConfigurationPreflightResult:
+    if fragment.fragment_type != WORKFLOW_DEFINITIONS_FRAGMENT:
+        return ConfigurationPreflightResult(
+            diagnostics=(_unsupported_fragment(fragment),),
+        )
+    diagnostics: list[ConfigurationDiagnostic] = []
+    plan: list[ConfigurationPlanItem] = []
+    try:
+        contributions = _contributions(fragment, context)
+    except (TypeError, ValueError) as exc:
+        return ConfigurationPreflightResult(
+            diagnostics=(
+                ConfigurationDiagnostic(
+                    severity="blocker",
+                    code="workflow_fragment_invalid",
+                    message=str(exc),
+                    module_id="workflow_engine",
+                    object_ref=fragment.fragment_id or fragment.fragment_type,
+                    resolution=(
+                        "Correct the workflow definition fragment before applying it."
+                    ),
+                ),
+            ),
+        )
+    for contribution in contributions:
+        graph = WorkflowGraph.model_validate(contribution.graph)
+        graph_diagnostics = validate_workflow_graph(graph)
+        errors = [item for item in graph_diagnostics if item.severity == "error"]
+        diagnostics.extend(
+            ConfigurationDiagnostic(
+                severity="blocker" if item.severity == "error" else "warning",
+                code=f"workflow_{item.code}",
+                message=item.message,
+                module_id="workflow_engine",
+                object_ref=(
+                    f"{contribution.definition_key}:{item.node_id or item.field or 'graph'}"
+                ),
+                resolution="Correct the workflow graph before applying it.",
+            )
+            for item in graph_diagnostics
+        )
+        tenant_id = context.tenant_id if contribution.scope_type == "tenant" else None
+        if contribution.scope_type == "tenant" and not tenant_id:
+            diagnostics.append(
+                ConfigurationDiagnostic(
+                    severity="blocker",
+                    code="workflow_tenant_required",
+                    message=(
+                        f"Workflow definition {contribution.definition_key!r} "
+                        "requires a tenant target."
+                    ),
+                    module_id="workflow_engine",
+                    object_ref=contribution.definition_key,
+                    resolution="Choose a tenant before applying this package.",
+                )
+            )
+        existing = _standard_definition(
+            session,
+            contribution=contribution,
+            tenant_id=tenant_id,
+        )
+        contribution_hash = workflow_definition_contribution_hash(contribution)
+        action = (
+            "create"
+            if existing is None
+            else (
+                "noop"
+                if existing.standard_contribution_hash == contribution_hash
+                else "update"
+            )
+        )
+        if errors or (contribution.scope_type == "tenant" and not tenant_id):
+            action = "blocked"
+        plan.append(
+            ConfigurationPlanItem(
+                action=action,
+                module_id="workflow_engine",
+                fragment_type=WORKFLOW_DEFINITIONS_FRAGMENT,
+                fragment_id=contribution.definition_key,
+                summary=(
+                    f"{action.capitalize()} module workflow baseline "
+                    f"{contribution.definition_key}."
+                ),
+            )
+        )
+    return ConfigurationPreflightResult(
+        diagnostics=tuple(diagnostics),
+        plan=tuple(plan),
+    )
+
+
+def apply_workflow_definitions(
+    session: Session,
+    fragment: ConfigurationPackageFragment,
+    context: ConfigurationPreflightContext,
+    *,
+    registry: object | None,
+) -> ConfigurationApplyResult:
+    preflight = preflight_workflow_definitions(session, fragment, context)
+    blockers = tuple(
+        item for item in preflight.diagnostics if item.severity == "blocker"
+    )
+    if blockers:
+        return ConfigurationApplyResult(diagnostics=blockers)
+    contributions = _contributions(fragment, context)
+    result = reconcile_workflow_contributions(
+        session,
+        contributions=contributions,
+        registry=registry,
+        tenant_ids=((context.tenant_id,) if context.tenant_id else ()),
+    )
+    diagnostics = [
+        ConfigurationDiagnostic(
+            severity="blocker",
+            code="workflow_contribution_blocked",
+            message=str(item.get("message") or "Workflow baseline was blocked."),
+            module_id="workflow_engine",
+            object_ref=str(item.get("definition_key") or "workflow_definition"),
+            resolution=(
+                "Install the required capabilities and correct the definition graph."
+            ),
+        )
+        for item in result["items"]
+        if isinstance(item, Mapping) and item.get("state") == "blocked"
+    ]
+    created: dict[str, str] = {}
+    updated: dict[str, str] = {}
+    for item in result["items"]:
+        if not isinstance(item, Mapping):
+            continue
+        definition_id = item.get("definition_id")
+        definition_key = item.get("definition_key")
+        if not definition_id or not definition_key:
+            continue
+        ref = f"workflow_definition:{definition_id}"
+        if item.get("state") == "created":
+            created[str(definition_key)] = ref
+        elif item.get("state") == "updated":
+            updated[str(definition_key)] = ref
+    return ConfigurationApplyResult(
+        diagnostics=tuple(diagnostics),
+        created_refs=created,
+        updated_refs=updated,
+    )
+
+
+def export_workflow_definitions(
+    session: Session,
+    selection: ConfigurationExportSelection,
+) -> ConfigurationExportResult:
+    query = select(WorkflowDefinition).where(
+        WorkflowDefinition.deleted_at.is_(None),
+        WorkflowDefinition.standard_origin_module_id.is_not(None),
+    )
+    if selection.tenant_id:
+        query = query.where(
+            WorkflowDefinition.tenant_id == selection.tenant_id,
+        )
+    elif "system" in selection.scopes:
+        query = query.where(WorkflowDefinition.tenant_id.is_(None))
+    else:
+        return ConfigurationExportResult(
+            diagnostics=(
+                ConfigurationDiagnostic(
+                    severity="blocker",
+                    code="workflow_export_scope_required",
+                    message=("Workflow export needs a tenant or the system scope."),
+                    module_id="workflow_engine",
+                    resolution="Choose a tenant or system scope before export.",
+                ),
+            ),
+        )
+    definitions = session.scalars(
+        query.order_by(WorkflowDefinition.definition_key.asc())
+    ).all()
+    items = [
+        _export_item(definition)
+        for definition in definitions
+        if not selection.object_refs
+        or definition.id in selection.object_refs
+        or f"workflow_definition:{definition.id}" in selection.object_refs
+    ]
+    return ConfigurationExportResult(
+        fragments=(
+            ConfigurationPackageFragment(
+                module_id="workflow_engine",
+                fragment_type=WORKFLOW_DEFINITIONS_FRAGMENT,
+                fragment_id="workflow-standards",
+                payload={"schema_version": 1, "items": items},
+            ),
+        ),
+    )
+
+
+def _contributions(
+    fragment: ConfigurationPackageFragment,
+    context: ConfigurationPreflightContext,
+) -> tuple[WorkflowDefinitionContribution, ...]:
+    raw_items = fragment.payload.get("items")
+    if not isinstance(raw_items, list):
+        raise ValueError("Workflow definitions fragment requires an items list.")
+    contributions: list[WorkflowDefinitionContribution] = []
+    for index, raw in enumerate(raw_items):
+        if not isinstance(raw, Mapping):
+            raise ValueError(f"Workflow definition item {index} must be an object.")
+        definition_key = _required_text(raw, "definition_key")
+        scope_type = str(raw.get("scope_type") or "tenant").strip().casefold()
+        if scope_type not in {"system", "tenant"}:
+            raise ValueError(
+                f"Workflow definition {definition_key!r} has unsupported scope "
+                f"{scope_type!r}."
+            )
+        graph = raw.get("graph")
+        if not isinstance(graph, Mapping):
+            raise ValueError(
+                f"Workflow definition {definition_key!r} requires a graph object."
+            )
+        origin_module_id = str(
+            raw.get("origin_module_id")
+            or fragment.payload.get("origin_module_id")
+            or "configuration_package"
+        ).strip()
+        origin_module_version = str(
+            raw.get("origin_module_version")
+            or fragment.payload.get("origin_module_version")
+            or "1"
+        ).strip()
+        contributions.append(
+            WorkflowDefinitionContribution(
+                origin_module_id=origin_module_id,
+                origin_module_version=origin_module_version,
+                definition_key=definition_key,
+                name=_required_text(raw, "name"),
+                description=_optional_text(raw.get("description")),
+                graph=dict(graph),
+                contribution_schema_version=str(
+                    raw.get("contribution_schema_version") or "1"
+                ),
+                definition_kind=str(raw.get("definition_kind") or "flow"),
+                scope_type=scope_type,
+                inherit_to_lower_scopes=_bool(
+                    raw.get("inherit_to_lower_scopes"),
+                    default=scope_type == "system",
+                ),
+                allow_start=_bool(raw.get("allow_start"), default=True),
+                allow_reuse=_bool(raw.get("allow_reuse"), default=True),
+                allow_automation=_bool(
+                    raw.get("allow_automation"),
+                    default=False,
+                ),
+                execution_mode=str(raw.get("execution_mode") or "hybrid"),
+                view_id=_optional_text(raw.get("view_id")),
+                view_revision_id=_optional_text(raw.get("view_revision_id")),
+                bpmn_xml=_optional_text(raw.get("bpmn_xml")),
+                bpmn_adapter_id=str(
+                    raw.get("bpmn_adapter_id") or "govoplan.native.bpmn"
+                ),
+                bpmn_adapter_version=_optional_text(raw.get("bpmn_adapter_version")),
+                activate_on_install=_bool(
+                    raw.get("activate_on_install"),
+                    default=True,
+                ),
+                required_capabilities=_strings(raw.get("required_capabilities")),
+                required_interfaces=_strings(raw.get("required_interfaces")),
+                metadata={
+                    **_mapping(raw.get("metadata")),
+                    "configuration_fragment_id": fragment.fragment_id,
+                    "configuration_tenant_id": context.tenant_id,
+                },
+                policy_metadata=_mapping(raw.get("policy_metadata")),
+            )
+        )
+    return tuple(contributions)
+
+
+def _standard_definition(
+    session: Session,
+    *,
+    contribution: WorkflowDefinitionContribution,
+    tenant_id: str | None,
+) -> WorkflowDefinition | None:
+    scope_key = "system" if tenant_id is None else f"tenant:{tenant_id}"
+    return session.scalar(
+        select(WorkflowDefinition).where(
+            WorkflowDefinition.scope_key == scope_key,
+            WorkflowDefinition.standard_origin_module_id
+            == contribution.origin_module_id,
+            WorkflowDefinition.standard_definition_key == contribution.definition_key,
+        )
+    )
+
+
+def _export_item(definition: WorkflowDefinition) -> dict[str, object]:
+    revision = next(
+        item
+        for item in definition.revisions
+        if item.revision == definition.current_revision
+    )
+    return {
+        "origin_module_id": definition.standard_origin_module_id,
+        "origin_module_version": definition.standard_origin_module_version,
+        "contribution_schema_version": (
+            definition.standard_contribution_schema_version or "1"
+        ),
+        "definition_key": (
+            definition.standard_definition_key or definition.definition_key
+        ),
+        "name": definition.name,
+        "description": definition.description,
+        "graph": dict(revision.graph),
+        "definition_kind": definition.definition_kind,
+        "scope_type": definition.scope_type,
+        "inherit_to_lower_scopes": definition.inherit_to_lower_scopes,
+        "allow_start": definition.allow_start,
+        "allow_reuse": definition.allow_reuse,
+        "allow_automation": definition.allow_automation,
+        "execution_mode": revision.execution_mode,
+        "view_id": revision.view_id,
+        "view_revision_id": revision.view_revision_id,
+        "bpmn_xml": revision.bpmn_xml,
+        "bpmn_adapter_id": revision.bpmn_adapter_id,
+        "bpmn_adapter_version": revision.bpmn_adapter_version,
+        "activate_on_install": definition.active_revision is not None,
+        "metadata": dict(definition.metadata_),
+    }
+
+
+def _unsupported_fragment(
+    fragment: ConfigurationPackageFragment,
+) -> ConfigurationDiagnostic:
+    return ConfigurationDiagnostic(
+        severity="blocker",
+        code="workflow_fragment_unsupported",
+        message=(
+            f"Workflow Engine does not support fragment type "
+            f"{fragment.fragment_type!r}."
+        ),
+        module_id="workflow_engine",
+        object_ref=fragment.fragment_id or fragment.fragment_type,
+        resolution="Use a workflow_definitions fragment.",
+    )
+
+
+def _required_text(value: Mapping[str, Any], key: str) -> str:
+    raw = value.get(key)
+    text = str(raw).strip() if raw is not None else ""
+    if not text:
+        raise ValueError(f"Workflow definition requires {key!r}.")
+    return text
+
+
+def _optional_text(value: object) -> str | None:
+    text = str(value).strip() if value is not None else ""
+    return text or None
+
+
+def _mapping(value: object) -> dict[str, Any]:
+    return dict(value) if isinstance(value, Mapping) else {}
+
+
+def _strings(value: object) -> tuple[str, ...]:
+    if value is None:
+        return ()
+    if not isinstance(value, (list, tuple)):
+        raise ValueError("Workflow requirement lists must be arrays.")
+    return tuple(str(item).strip() for item in value if str(item).strip())
+
+
+def _bool(value: object, *, default: bool) -> bool:
+    return value if isinstance(value, bool) else default
+
+
+__all__ = [
+    "SqlWorkflowConfigurationProvider",
+    "WORKFLOW_CONFIGURATION_CAPABILITY",
+    "WORKFLOW_DEFINITIONS_FRAGMENT",
+    "apply_workflow_definitions",
+    "export_workflow_definitions",
+    "preflight_workflow_definitions",
+]

src/govoplan_workflow_engine/backend/contributions.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
-from sqlalchemy import select
+from sqlalchemy import inspect, select
 from sqlalchemy.orm import Session
 
 from govoplan_core.core.workflows import (
@@ -18,7 +18,10 @@ from govoplan_workflow_engine.backend.db.models import (
 from govoplan_workflow_engine.backend.schemas import (
     BpmnRevisionInput,
     WorkflowGraph,
+    WorkflowStandardDiffItemResponse,
+    WorkflowStandardDiffResponse,
 )
+from govoplan_workflow_engine.backend.bpmn_graph import canonical_bpmn_graph
 from govoplan_workflow_engine.backend.service import (
     WorkflowConflictError,
     WorkflowNotFoundError,
@@ -29,6 +32,7 @@ from govoplan_workflow_engine.backend.service import (
     _validate_bpmn_activation,
     _validate_execution_mode_activation,
     get_definition,
+    get_definition_revision,
 )
 
 
@@ -38,7 +42,21 @@ def reconcile_workflow_definition_contributions(
     registry: object | None,
     tenant_ids: Sequence[str] = (),
 ) -> dict[str, object]:
-    contributions = _registered_contributions(registry)
+    return reconcile_workflow_contributions(
+        session,
+        contributions=_registered_contributions(registry),
+        registry=registry,
+        tenant_ids=tenant_ids,
+    )
+
+
+def reconcile_workflow_contributions(
+    session: Session,
+    *,
+    contributions: Sequence[WorkflowDefinitionContribution],
+    registry: object | None,
+    tenant_ids: Sequence[str] = (),
+) -> dict[str, object]:
     normalized_tenant_ids = tuple(
         dict.fromkeys(str(item).strip() for item in tenant_ids if str(item).strip())
     )
@@ -163,6 +181,199 @@ def reset_workflow_override_to_standard(
     return baseline
 
 
+def compare_workflow_override_to_standard(
+    session: Session,
+    *,
+    tenant_id: str,
+    definition_id: str,
+    include_unchanged: bool = False,
+) -> WorkflowStandardDiffResponse:
+    override = get_definition(
+        session,
+        tenant_id=tenant_id,
+        definition_id=definition_id,
+    )
+    if not override.derived_from_definition_id:
+        raise WorkflowConflictError(
+            "Only a definition derived from a module standard can be compared."
+        )
+    baseline = session.get(
+        WorkflowDefinition,
+        override.derived_from_definition_id,
+    )
+    if (
+        baseline is None
+        or baseline.deleted_at is not None
+        or not baseline.standard_origin_module_id
+    ):
+        raise WorkflowNotFoundError(
+            "The module standard behind this override is unavailable."
+        )
+    pinned_revision_number = override.derived_from_revision
+    if pinned_revision_number is None:
+        raise WorkflowConflictError(
+            "The override does not pin a module standard revision."
+        )
+    pinned = get_definition_revision(
+        session,
+        definition=baseline,
+        revision=pinned_revision_number,
+    )
+    local = get_definition_revision(
+        session,
+        definition=override,
+        revision=override.current_revision,
+    )
+    latest = get_definition_revision(
+        session,
+        definition=baseline,
+        revision=baseline.current_revision,
+    )
+    items = _semantic_graph_diff(
+        _canonical_graph_payload(pinned.graph),
+        _canonical_graph_payload(local.graph),
+        _canonical_graph_payload(latest.graph),
+    )
+    counts = {
+        state: sum(item.state == state for item in items)
+        for state in (
+            "unchanged",
+            "local_only",
+            "upstream_only",
+            "same_change",
+            "conflict",
+        )
+    }
+    visible_items = (
+        items
+        if include_unchanged
+        else [item for item in items if item.state != "unchanged"]
+    )
+    return WorkflowStandardDiffResponse(
+        override_definition_id=override.id,
+        baseline_definition_id=baseline.id,
+        pinned_baseline_revision=pinned.revision,
+        local_revision=local.revision,
+        latest_baseline_revision=latest.revision,
+        counts=counts,
+        conflict_count=counts["conflict"],
+        auto_mergeable=counts["conflict"] == 0,
+        items=visible_items,
+    )
+
+
+def _canonical_graph_payload(raw: Mapping[str, object]) -> dict[str, object]:
+    return canonical_bpmn_graph(WorkflowGraph.model_validate(raw)).model_dump(
+        mode="json"
+    )
+
+
+def _semantic_graph_diff(
+    baseline: Mapping[str, object],
+    local: Mapping[str, object],
+    latest: Mapping[str, object],
+) -> list[WorkflowStandardDiffItemResponse]:
+    items = [
+        _diff_item(
+            resource_type="graph",
+            resource_id="graph",
+            baseline={
+                "schema_version": baseline.get("schema_version"),
+                "metadata": baseline.get("metadata", {}),
+            },
+            local={
+                "schema_version": local.get("schema_version"),
+                "metadata": local.get("metadata", {}),
+            },
+            latest={
+                "schema_version": latest.get("schema_version"),
+                "metadata": latest.get("metadata", {}),
+            },
+        )
+    ]
+    for resource_type, collection_name in (
+        ("node", "nodes"),
+        ("edge", "edges"),
+    ):
+        baseline_items = _items_by_id(baseline.get(collection_name))
+        local_items = _items_by_id(local.get(collection_name))
+        latest_items = _items_by_id(latest.get(collection_name))
+        for resource_id in sorted(
+            set(baseline_items) | set(local_items) | set(latest_items)
+        ):
+            items.append(
+                _diff_item(
+                    resource_type=resource_type,
+                    resource_id=resource_id,
+                    baseline=baseline_items.get(resource_id),
+                    local=local_items.get(resource_id),
+                    latest=latest_items.get(resource_id),
+                )
+            )
+    return items
+
+
+def _items_by_id(value: object) -> dict[str, dict[str, object]]:
+    if not isinstance(value, list):
+        return {}
+    return {
+        str(item["id"]): dict(item)
+        for item in value
+        if isinstance(item, Mapping) and item.get("id") is not None
+    }
+
+
+def _diff_item(
+    *,
+    resource_type: str,
+    resource_id: str,
+    baseline: dict[str, object] | None,
+    local: dict[str, object] | None,
+    latest: dict[str, object] | None,
+) -> WorkflowStandardDiffItemResponse:
+    local_changed = local != baseline
+    upstream_changed = latest != baseline
+    if not local_changed and not upstream_changed:
+        state = "unchanged"
+        action = "none"
+    elif local_changed and not upstream_changed:
+        state = "local_only"
+        action = "keep_local"
+    elif upstream_changed and not local_changed:
+        state = "upstream_only"
+        action = "adopt_upstream"
+    elif local == latest:
+        state = "same_change"
+        action = "either"
+    else:
+        state = "conflict"
+        action = "manual_resolution"
+    return WorkflowStandardDiffItemResponse(
+        resource_type=resource_type,
+        resource_id=resource_id,
+        state=state,
+        recommended_action=action,
+        changed_fields=_changed_fields(baseline, local, latest),
+        baseline=baseline,
+        local=local,
+        latest=latest,
+    )
+
+
+def _changed_fields(
+    baseline: Mapping[str, object] | None,
+    local: Mapping[str, object] | None,
+    latest: Mapping[str, object] | None,
+) -> list[str]:
+    if baseline is None or local is None or latest is None:
+        return ["existence"]
+    return sorted(
+        key
+        for key in set(baseline) | set(local) | set(latest)
+        if not (baseline.get(key) == local.get(key) == latest.get(key))
+    )
+
+
 class SqlWorkflowDefinitionContributionProvider:
     def __init__(self, *, registry: object | None = None) -> None:
         self._registry = registry
@@ -175,6 +386,20 @@ class SqlWorkflowDefinitionContributionProvider:
     ) -> Mapping[str, object]:
         if not isinstance(session, Session):
             raise TypeError("Workflow contribution reconciliation requires a Session.")
+        inspector = inspect(session.get_bind())
+        required_tables = {
+            WorkflowDefinition.__tablename__,
+            WorkflowDefinitionRevision.__tablename__,
+        }
+        missing_tables = sorted(
+            table for table in required_tables if not inspector.has_table(table)
+        )
+        if missing_tables:
+            return {
+                "skipped": True,
+                "reason": "schema_unavailable",
+                "missing_tables": missing_tables,
+            }
         return reconcile_workflow_definition_contributions(
             session,
             registry=self._registry,
@@ -464,6 +689,8 @@ def _item(
 
 __all__ = [
     "SqlWorkflowDefinitionContributionProvider",
+    "compare_workflow_override_to_standard",
+    "reconcile_workflow_contributions",
     "reconcile_workflow_definition_contributions",
     "reset_workflow_override_to_standard",
 ]

src/govoplan_workflow_engine/backend/manifest.py

@@ -33,10 +33,14 @@ from govoplan_core.core.references import CAPABILITY_ACCESS_REFERENCE_OPTIONS
 from govoplan_core.core.views import CAPABILITY_VIEWS_RESOLVER
 from govoplan_core.core.workflows import (
     CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS,
+    CAPABILITY_WORKFLOW_ORCHESTRATION,
     CAPABILITY_WORKFLOW_RUNTIME_WORKER,
 )
 from govoplan_core.db.base import Base
 from govoplan_workflow_engine.backend.db import models as workflow_models
+from govoplan_workflow_engine.backend.configuration_provider import (
+    WORKFLOW_CONFIGURATION_CAPABILITY,
+)
 
 
 MODULE_ID = "workflow_engine"
@@ -144,6 +148,22 @@ def _definition_contribution_provider(context: ModuleContext):
     return SqlWorkflowDefinitionContributionProvider(registry=context.registry)
 
 
+def _configuration_provider(context: ModuleContext):
+    from govoplan_workflow_engine.backend.configuration_provider import (
+        SqlWorkflowConfigurationProvider,
+    )
+
+    return SqlWorkflowConfigurationProvider(registry=context.registry)
+
+
+def _orchestration_provider(context: ModuleContext):
+    from govoplan_workflow_engine.backend.orchestration import (
+        SqlWorkflowOrchestrationProvider,
+    )
+
+    return SqlWorkflowOrchestrationProvider(registry=context.registry)
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
@@ -179,6 +199,14 @@ manifest = ModuleManifest(
             name="workflow.definition_contributions",
             version="0.1.0",
         ),
+        ModuleInterfaceProvider(
+            name=WORKFLOW_CONFIGURATION_CAPABILITY,
+            version="1.0.0",
+        ),
+        ModuleInterfaceProvider(
+            name=CAPABILITY_WORKFLOW_ORCHESTRATION,
+            version="1.0.0",
+        ),
         ModuleInterfaceProvider(name="workflow.runtime_worker", version=MODULE_VERSION),
         ModuleInterfaceProvider(name="workflow.bpmn_interchange", version="1.0.0"),
         ModuleInterfaceProvider(
@@ -232,6 +260,8 @@ manifest = ModuleManifest(
             _definition_contribution_provider
         ),
         CAPABILITY_WORKFLOW_RUNTIME_WORKER: _runtime_worker,
+        CAPABILITY_WORKFLOW_ORCHESTRATION: _orchestration_provider,
+        WORKFLOW_CONFIGURATION_CAPABILITY: _configuration_provider,
     },
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,

src/govoplan_workflow_engine/backend/orchestration.py

@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+from sqlalchemy import case, or_, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.workflows import (
+    WorkflowCurrentStepResolution,
+    WorkflowInstanceRef,
+    WorkflowStandardStartRequest,
+)
+from govoplan_workflow_engine.backend.db.models import (
+    WorkflowDefinition,
+    WorkflowDefinitionRevision,
+    WorkflowInstance,
+    WorkflowInstanceStep,
+)
+from govoplan_workflow_engine.backend.instance_service import (
+    get_instance as get_workflow_instance,
+    resolve_step,
+    start_instance,
+)
+from govoplan_workflow_engine.backend.schemas import (
+    WorkflowInstanceStartRequest,
+    WorkflowStepActionRequest,
+)
+
+
+class SqlWorkflowOrchestrationProvider:
+    def __init__(self, *, registry: object | None = None) -> None:
+        self._registry = registry
+
+    def start_standard(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: WorkflowStandardStartRequest,
+    ) -> WorkflowInstanceRef:
+        sql_session = _session(session)
+        api_principal = _principal(principal)
+        definition = _active_standard(
+            sql_session,
+            tenant_id=request.tenant_id,
+            origin_module_id=request.origin_module_id,
+            definition_key=request.definition_key,
+        )
+        instance, replayed = start_instance(
+            sql_session,
+            tenant_id=request.tenant_id,
+            definition_id=definition.id,
+            actor_id=request.actor_id,
+            principal=api_principal,
+            registry=self._registry,
+            payload=WorkflowInstanceStartRequest(
+                idempotency_key=request.idempotency_key,
+                input=dict(request.input),
+                correlation_id=request.correlation_id,
+            ),
+            start_origin=request.start_origin,
+        )
+        return _instance_ref(sql_session, instance, replayed=replayed)
+
+    def resolve_current_step(
+        self,
+        session: object,
+        principal: object,
+        *,
+        tenant_id: str,
+        instance_id: str,
+        resolution: WorkflowCurrentStepResolution,
+    ) -> WorkflowInstanceRef:
+        sql_session = _session(session)
+        api_principal = _principal(principal)
+        instance = get_workflow_instance(
+            sql_session,
+            tenant_id=tenant_id,
+            instance_id=instance_id,
+        )
+        step_id = instance.current_step_id
+        if step_id is None:
+            raise ValueError("Workflow instance has no current waiting step.")
+        if (
+            resolution.expected_step_id is not None
+            and resolution.expected_step_id != step_id
+        ):
+            raise ValueError("Workflow current step changed; reload and retry.")
+        resolved = resolve_step(
+            sql_session,
+            tenant_id=tenant_id,
+            instance_id=instance_id,
+            step_id=step_id,
+            actor_id=resolution.actor_id,
+            principal=api_principal,
+            registry=self._registry,
+            payload=WorkflowStepActionRequest(
+                action=resolution.action,  # type: ignore[arg-type]
+                output=dict(resolution.output),
+                evidence=list(resolution.evidence),
+                comment=resolution.comment,
+            ),
+        )
+        return _instance_ref(sql_session, resolved)
+
+    def get_instance(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        instance_id: str,
+    ) -> WorkflowInstanceRef:
+        sql_session = _session(session)
+        instance = get_workflow_instance(
+            sql_session,
+            tenant_id=tenant_id,
+            instance_id=instance_id,
+        )
+        return _instance_ref(sql_session, instance)
+
+
+def _active_standard(
+    session: Session,
+    *,
+    tenant_id: str,
+    origin_module_id: str,
+    definition_key: str,
+) -> WorkflowDefinition:
+    definition = session.scalar(
+        select(WorkflowDefinition)
+        .where(
+            or_(
+                WorkflowDefinition.tenant_id == tenant_id,
+                WorkflowDefinition.tenant_id.is_(None),
+            ),
+            WorkflowDefinition.standard_origin_module_id == origin_module_id,
+            WorkflowDefinition.standard_definition_key == definition_key,
+            WorkflowDefinition.status == "active",
+            WorkflowDefinition.active_revision.is_not(None),
+            WorkflowDefinition.deleted_at.is_(None),
+        )
+        .order_by(
+            case(
+                (WorkflowDefinition.tenant_id == tenant_id, 0),
+                else_=1,
+            ),
+            WorkflowDefinition.updated_at.desc(),
+        )
+    )
+    if definition is None:
+        raise ValueError(
+            f"Active Workflow standard is unavailable: "
+            f"{origin_module_id}/{definition_key}."
+        )
+    return definition
+
+
+def _instance_ref(
+    session: Session,
+    instance: WorkflowInstance,
+    *,
+    replayed: bool = False,
+) -> WorkflowInstanceRef:
+    revision = session.get(
+        WorkflowDefinitionRevision,
+        instance.definition_revision_id,
+    )
+    if revision is None:
+        raise ValueError("Workflow instance revision is unavailable.")
+    step = (
+        session.get(WorkflowInstanceStep, instance.current_step_id)
+        if instance.current_step_id
+        else None
+    )
+    return WorkflowInstanceRef(
+        id=instance.id,
+        tenant_id=instance.tenant_id,
+        definition_id=instance.definition_id,
+        definition_revision_id=instance.definition_revision_id,
+        definition_revision=revision.revision,
+        definition_hash=revision.content_hash,
+        status=instance.status,
+        current_step_id=instance.current_step_id,
+        current_node_id=step.node_id if step is not None else None,
+        replayed=replayed,
+    )
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Workflow orchestration requires a SQLAlchemy Session.")
+    return value
+
+
+def _principal(value: object) -> ApiPrincipal:
+    if not isinstance(value, ApiPrincipal):
+        raise TypeError("Workflow orchestration requires an API principal.")
+    return value
+
+
+__all__ = ["SqlWorkflowOrchestrationProvider"]

src/govoplan_workflow_engine/backend/router.py

@@ -21,6 +21,7 @@ from govoplan_workflow_engine.backend.governance import (
     require_definition_action,
 )
 from govoplan_workflow_engine.backend.contributions import (
+    compare_workflow_override_to_standard,
     reconcile_workflow_definition_contributions,
     reset_workflow_override_to_standard,
 )
@@ -87,6 +88,7 @@ from govoplan_workflow_engine.backend.schemas import (
     WorkflowNodeLibraryResponse,
     WorkflowNodeTypeResponse,
     WorkflowPortResponse,
+    WorkflowStandardDiffResponse,
     WorkflowStepActionRequest,
 )
 from govoplan_workflow_engine.backend.instance_service import (
@@ -819,6 +821,41 @@ def api_reset_definition_to_standard(
     return response
 
 
+@router.get(
+    "/definitions/{definition_id}/standard-diff",
+    response_model=WorkflowStandardDiffResponse,
+)
+def api_compare_definition_to_standard(
+    definition_id: str,
+    include_unchanged: bool = False,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> WorkflowStandardDiffResponse:
+    _require_any_scope(principal, DEFINITION_READ_SCOPE, ADMIN_SCOPE)
+    try:
+        definition = get_definition(
+            session,
+            tenant_id=principal.tenant_id,
+            definition_id=definition_id,
+        )
+        require_definition_action(
+            definition,
+            principal=principal,
+            registry=get_registry(),
+            action="view",
+        )
+        return compare_workflow_override_to_standard(
+            session,
+            tenant_id=principal.tenant_id,
+            definition_id=definition_id,
+            include_unchanged=include_unchanged,
+        )
+    except PermissionError as exc:
+        raise _governance_http_error(exc) from exc
+    except WorkflowError as exc:
+        raise _http_error(exc) from exc
+
+
 @router.get(
     "/instances/{instance_id}",
     response_model=WorkflowInstanceResponse,

src/govoplan_workflow_engine/backend/schemas.py

@@ -309,6 +309,41 @@ class WorkflowStandardProvenanceResponse(BaseModel):
     reset_available: bool = False
 
 
+class WorkflowStandardDiffItemResponse(BaseModel):
+    resource_type: Literal["graph", "node", "edge"]
+    resource_id: str
+    state: Literal[
+        "unchanged",
+        "local_only",
+        "upstream_only",
+        "same_change",
+        "conflict",
+    ]
+    recommended_action: Literal[
+        "none",
+        "keep_local",
+        "adopt_upstream",
+        "either",
+        "manual_resolution",
+    ]
+    changed_fields: list[str] = Field(default_factory=list)
+    baseline: dict[str, Any] | None = None
+    local: dict[str, Any] | None = None
+    latest: dict[str, Any] | None = None
+
+
+class WorkflowStandardDiffResponse(BaseModel):
+    override_definition_id: str
+    baseline_definition_id: str
+    pinned_baseline_revision: int
+    local_revision: int
+    latest_baseline_revision: int
+    counts: dict[str, int]
+    conflict_count: int
+    auto_mergeable: bool
+    items: list[WorkflowStandardDiffItemResponse]
+
+
 class WorkflowActionDecisionResponse(BaseModel):
     allowed: bool
     reason: str | None = None

tests/test_configuration_provider.py

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.configuration_packages import (
+    ConfigurationExportSelection,
+    ConfigurationPackageFragment,
+    ConfigurationPreflightContext,
+)
+from govoplan_core.db.base import Base
+from govoplan_workflow_engine.backend.configuration_provider import (
+    WORKFLOW_DEFINITIONS_FRAGMENT,
+    apply_workflow_definitions,
+    export_workflow_definitions,
+    preflight_workflow_definitions,
+)
+from govoplan_workflow_engine.backend.db.models import (
+    WorkflowDefinition,
+    WorkflowDefinitionRevision,
+)
+
+
+def fragment(*, label: str = "Review request") -> ConfigurationPackageFragment:
+    return ConfigurationPackageFragment(
+        module_id="workflow_engine",
+        fragment_type=WORKFLOW_DEFINITIONS_FRAGMENT,
+        fragment_id="permit-workflows",
+        payload={
+            "origin_module_id": "permits",
+            "origin_module_version": "1.0.0",
+            "items": [
+                {
+                    "definition_key": "permit-review",
+                    "name": "Permit review",
+                    "scope_type": "system",
+                    "graph": {
+                        "schema_version": 1,
+                        "nodes": [
+                            {
+                                "id": "start",
+                                "type": "workflow.start.manual",
+                                "label": "Start",
+                                "config": {"input_schema_ref": ""},
+                            },
+                            {
+                                "id": "review",
+                                "type": "workflow.activity",
+                                "label": label,
+                                "config": {
+                                    "title": label,
+                                    "instructions": "",
+                                    "assignee": "",
+                                    "due_after": "",
+                                },
+                            },
+                            {
+                                "id": "done",
+                                "type": "workflow.end.completed",
+                                "label": "Done",
+                                "config": {"output_mapping": {}},
+                            },
+                        ],
+                        "edges": [
+                            {
+                                "id": "start-review",
+                                "source": "start",
+                                "target": "review",
+                            },
+                            {"id": "review-done", "source": "review", "target": "done"},
+                        ],
+                    },
+                },
+            ],
+        },
+    )
+
+
+class WorkflowConfigurationProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[
+                WorkflowDefinition.__table__,
+                WorkflowDefinitionRevision.__table__,
+            ],
+        )
+        self.Session = sessionmaker(bind=self.engine)
+        self.session: Session = self.Session()
+        self.context = ConfigurationPreflightContext(
+            installed_modules={"workflow_engine": "0.1.14"},
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[
+                WorkflowDefinitionRevision.__table__,
+                WorkflowDefinition.__table__,
+            ],
+        )
+        self.engine.dispose()
+
+    def test_fragment_preflight_apply_update_and_export(self) -> None:
+        initial = preflight_workflow_definitions(
+            self.session,
+            fragment(),
+            self.context,
+        )
+        created = apply_workflow_definitions(
+            self.session,
+            fragment(),
+            self.context,
+            registry=None,
+        )
+        repeated = apply_workflow_definitions(
+            self.session,
+            fragment(),
+            self.context,
+            registry=None,
+        )
+        updated = apply_workflow_definitions(
+            self.session,
+            fragment(label="Review corrected request"),
+            self.context,
+            registry=None,
+        )
+        exported = export_workflow_definitions(
+            self.session,
+            ConfigurationExportSelection(scopes=("system",)),
+        )
+
+        definition = self.session.scalar(select(WorkflowDefinition))
+        assert definition is not None
+        self.assertEqual("create", initial.plan[0].action)
+        self.assertIn("permit-review", created.created_refs)
+        self.assertFalse(repeated.created_refs)
+        self.assertFalse(repeated.updated_refs)
+        self.assertIn("permit-review", updated.updated_refs)
+        self.assertEqual(2, definition.current_revision)
+        self.assertEqual("workflow_definitions", exported.fragments[0].fragment_type)
+        self.assertEqual(
+            "permit-review",
+            exported.fragments[0].payload["items"][0]["definition_key"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_contributions.py

@@ -11,9 +11,15 @@ from govoplan_core.core.access import PrincipalRef
 from govoplan_core.core.modules import ModuleManifest
 from govoplan_core.core.policy import PolicyDecision
 from govoplan_core.core.registry import PlatformRegistry
-from govoplan_core.core.workflows import WorkflowDefinitionContribution
+from govoplan_core.core.workflows import (
+    WorkflowCurrentStepResolution,
+    WorkflowDefinitionContribution,
+    WorkflowStandardStartRequest,
+)
 from govoplan_core.db.base import Base
 from govoplan_workflow_engine.backend.contributions import (
+    SqlWorkflowDefinitionContributionProvider,
+    compare_workflow_override_to_standard,
     reconcile_workflow_definition_contributions,
     reset_workflow_override_to_standard,
 )
@@ -25,6 +31,9 @@ from govoplan_workflow_engine.backend.db.models import (
     WorkflowInstanceStep,
 )
 from govoplan_workflow_engine.backend.instance_service import start_instance
+from govoplan_workflow_engine.backend.orchestration import (
+    SqlWorkflowOrchestrationProvider,
+)
 from govoplan_workflow_engine.backend.schemas import (
     WorkflowDefinitionDeriveRequest,
     WorkflowDefinitionUpdateRequest,
@@ -129,8 +138,12 @@ class _DefinitionPolicy:
     def resolve_definition_action(self, *, request):
         return PolicyDecision(
             allowed=(
-                request.action in {"view", "derive", "reuse"}
+                (request.action in {"view", "derive", "reuse"} and request.allow_reuse)
-                and request.allow_reuse
+                or (
+                    request.action == "run"
+                    and request.allow_run
+                    and request.status == "active"
+                )
             ),
             reason=None,
         )
@@ -209,6 +222,17 @@ class WorkflowContributionTests(unittest.TestCase):
             definition.revisions[1].contribution_hash,
         )
 
+    def test_provider_defers_until_engine_schema_is_available(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        with Session(engine) as empty_session:
+            result = SqlWorkflowDefinitionContributionProvider(
+                registry=registry_for(contribution())
+            ).reconcile(empty_session)
+        engine.dispose()
+
+        self.assertTrue(result["skipped"])
+        self.assertEqual("schema_unavailable", result["reason"])
+
     def test_standard_is_immutable_and_reset_archives_only_the_override(self) -> None:
         reconcile_workflow_definition_contributions(
             self.session,
@@ -300,6 +324,61 @@ class WorkflowContributionTests(unittest.TestCase):
             result["items"][0]["missing"],  # type: ignore[index]
         )
 
+    def test_semantic_three_way_diff_identifies_overlapping_node_changes(self) -> None:
+        reconcile_workflow_definition_contributions(
+            self.session,
+            registry=registry_for(contribution()),
+        )
+        baseline = self.session.scalar(select(WorkflowDefinition))
+        assert baseline is not None
+        override = derive_definition(
+            self.session,
+            tenant_id="tenant-1",
+            actor_id="account-1",
+            principal=principal(),
+            registry=_PolicyRegistry(),
+            source_definition_id=baseline.id,
+            payload=WorkflowDefinitionDeriveRequest(
+                name="Tenant approval override",
+                scope_type="tenant",
+            ),
+        )
+        local_graph = contribution_graph(label="Local approval wording")
+        update_definition(
+            self.session,
+            tenant_id="tenant-1",
+            definition_id=override.id,
+            actor_id="account-1",
+            payload=WorkflowDefinitionUpdateRequest(
+                name=override.name,
+                graph=local_graph,
+                expected_revision=override.current_revision,
+            ),
+        )
+        reconcile_workflow_definition_contributions(
+            self.session,
+            registry=registry_for(
+                contribution(version="1.1.0", label="Upstream approval wording")
+            ),
+        )
+
+        comparison = compare_workflow_override_to_standard(
+            self.session,
+            tenant_id="tenant-1",
+            definition_id=override.id,
+        )
+
+        review = next(
+            item
+            for item in comparison.items
+            if item.resource_type == "node" and item.resource_id == "review"
+        )
+        self.assertEqual("conflict", review.state)
+        self.assertEqual("manual_resolution", review.recommended_action)
+        self.assertIn("label", review.changed_fields)
+        self.assertEqual(1, comparison.conflict_count)
+        self.assertFalse(comparison.auto_mergeable)
+
     def test_domain_contribution_runs_without_the_editor_module(self) -> None:
         item = replace(
             contribution(),
@@ -333,6 +412,42 @@ class WorkflowContributionTests(unittest.TestCase):
         self.assertEqual("waiting", instance.status)
         self.assertEqual(definition.revisions[0].id, instance.definition_revision_id)
 
+    def test_headless_orchestration_starts_and_advances_a_standard(self) -> None:
+        registry = registry_for(contribution())
+        reconcile_workflow_definition_contributions(
+            self.session,
+            registry=registry,
+        )
+        provider = SqlWorkflowOrchestrationProvider(registry=_PolicyRegistry())
+
+        started = provider.start_standard(
+            self.session,
+            principal(),
+            request=WorkflowStandardStartRequest(
+                tenant_id="tenant-1",
+                origin_module_id="permits",
+                definition_key="application-approval",
+                idempotency_key="approval-1",
+                actor_id="account-1",
+                input={"request_id": "request-1"},
+            ),
+        )
+        completed = provider.resolve_current_step(
+            self.session,
+            principal(),
+            tenant_id="tenant-1",
+            instance_id=started.id,
+            resolution=WorkflowCurrentStepResolution(
+                action="complete",
+                expected_step_id=started.current_step_id,
+                actor_id="account-1",
+            ),
+        )
+
+        self.assertEqual("review", started.current_node_id)
+        self.assertEqual("completed", completed.status)
+        self.assertIsNone(completed.current_step_id)
+
 
 if __name__ == "__main__":
     unittest.main()