diff --git a/src/govoplan_workflow_engine/backend/configuration_provider.py b/src/govoplan_workflow_engine/backend/configuration_provider.py new file mode 100644 index 0000000..f3e57eb --- /dev/null +++ b/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", +] diff --git a/src/govoplan_workflow_engine/backend/contributions.py b/src/govoplan_workflow_engine/backend/contributions.py index 436d8ba..6e17ba6 100644 --- a/src/govoplan_workflow_engine/backend/contributions.py +++ b/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", ] diff --git a/src/govoplan_workflow_engine/backend/manifest.py b/src/govoplan_workflow_engine/backend/manifest.py index 935609b..1ebcaa2 100644 --- a/src/govoplan_workflow_engine/backend/manifest.py +++ b/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, diff --git a/src/govoplan_workflow_engine/backend/orchestration.py b/src/govoplan_workflow_engine/backend/orchestration.py new file mode 100644 index 0000000..b4aec90 --- /dev/null +++ b/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"] diff --git a/src/govoplan_workflow_engine/backend/router.py b/src/govoplan_workflow_engine/backend/router.py index f46681f..80c8b8e 100644 --- a/src/govoplan_workflow_engine/backend/router.py +++ b/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, diff --git a/src/govoplan_workflow_engine/backend/schemas.py b/src/govoplan_workflow_engine/backend/schemas.py index b9061d9..cd92fe2 100644 --- a/src/govoplan_workflow_engine/backend/schemas.py +++ b/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 diff --git a/tests/test_configuration_provider.py b/tests/test_configuration_provider.py new file mode 100644 index 0000000..334fa31 --- /dev/null +++ b/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() diff --git a/tests/test_contributions.py b/tests/test_contributions.py index 41c7299..544e712 100644 --- a/tests/test_contributions.py +++ b/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"} - and request.allow_reuse + (request.action in {"view", "derive", "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()