from __future__ import annotations from collections.abc import Mapping from datetime import datetime import hashlib import logging from sqlalchemy import func, select from sqlalchemy.orm import Session, selectinload from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.automation import ( ActionDefinition, ActionExecutionRequest, ActionExecutionResult, ActionPreview, AutomationInvocation, AutomationPrincipalRequest, action_effect_provider, automation_principal_provider, ) from govoplan_core.core.dataflows import ( DataflowPublicationTarget, DataflowRunDescriptor, DataflowRunRequest, dataflow_run_lifecycle, ) from govoplan_core.core.notifications import ( NotificationDispatchRequest, notification_dispatch_provider, ) from govoplan_core.core.events import ( EventActorRef, EventObjectRef, EventTenantRef, PlatformEvent, emit_platform_event, ) from govoplan_core.db.base import utcnow from govoplan_workflow_engine.backend.db.models import ( WorkflowDefinition, WorkflowDefinitionRevision, WorkflowInstance, WorkflowInstanceEvent, WorkflowInstanceStep, ) from govoplan_workflow_engine.backend.governance import require_definition_action from govoplan_workflow_engine.backend.bpmn_graph import ( BpmnGraphError, materialize_runtime_graph, ) from govoplan_workflow_engine.backend.schemas import ( WorkflowGraph, WorkflowInstanceEventResponse, WorkflowInstanceResponse, WorkflowInstanceStartRequest, WorkflowInstanceStepResponse, WorkflowNode, WorkflowStepActionRequest, WorkflowViewContextResponse, ) from govoplan_workflow_engine.backend.service import ( WorkflowConflictError, WorkflowNotFoundError, get_definition, get_definition_revision, ) from govoplan_workflow_engine.backend.recovery import ( WorkflowActionRecovery, WorkflowRecoveryBusy, WorkflowRecoveryConflict, WorkflowRecoveryError, acquire_workflow_state_fence, begin_workflow_action_recovery, canonical_sha256, claim_workflow_action_recovery, reconcile_stale_workflow_action_recovery, release_workflow_state_fence, ) INSTANCE_START_SCOPE = "workflow:instance:start" INSTANCE_TRANSITION_SCOPE = "workflow:instance:transition" DATAFLOW_RUN_SCOPE = "dataflow:pipeline:run" MAX_INSTANCE_TRANSITIONS = 100 logger = logging.getLogger(__name__) def list_instances( session: Session, *, tenant_id: str, definition_id: str | None = None, limit: int = 100, ) -> list[WorkflowInstance]: statement = ( select(WorkflowInstance) .where(WorkflowInstance.tenant_id == tenant_id) .options( selectinload(WorkflowInstance.steps), selectinload(WorkflowInstance.events), selectinload(WorkflowInstance.definition), ) .order_by( WorkflowInstance.updated_at.desc(), WorkflowInstance.id.desc(), ) .limit(max(1, min(int(limit), 200))) ) if definition_id: statement = statement.where(WorkflowInstance.definition_id == definition_id) return list(session.scalars(statement)) def get_instance( session: Session, *, tenant_id: str, instance_id: str, for_update: bool = False, ) -> WorkflowInstance: statement = ( select(WorkflowInstance) .where( WorkflowInstance.id == instance_id, WorkflowInstance.tenant_id == tenant_id, ) .options( selectinload(WorkflowInstance.steps), selectinload(WorkflowInstance.events), selectinload(WorkflowInstance.definition), ) ) if for_update: statement = statement.with_for_update() instance = session.scalar(statement) if instance is None: raise WorkflowNotFoundError("Workflow instance not found.") return instance def start_instance( session: Session, *, tenant_id: str, definition_id: str, actor_id: str | None, principal: ApiPrincipal, registry: object | None, payload: WorkflowInstanceStartRequest, start_origin: str = "user", ) -> tuple[WorkflowInstance, bool]: definition = get_definition( session, tenant_id=tenant_id, definition_id=definition_id, ) if definition.definition_kind == "template": raise WorkflowConflictError("Workflow templates cannot be started.") if definition.status != "active" or definition.active_revision is None: raise WorkflowConflictError( "Activate a Workflow revision before starting an instance." ) try: require_definition_action( definition, principal=principal, registry=registry, action="start", ) except PermissionError as exc: raise WorkflowConflictError(str(exc)) from exc revision = get_definition_revision( session, definition=definition, revision=definition.active_revision, ) normalized_origin = _normalize_start_origin(start_origin) if normalized_origin != "user" and not definition.allow_automation: raise WorkflowConflictError("This Workflow does not allow automated starts.") if revision.execution_mode == "guided" and normalized_origin != "user": raise WorkflowConflictError( "Guided workflows must be started by a user; use hybrid mode " "for triggered workflows with human handoffs." ) graph = _runtime_graph(revision) _require_runtime_dependencies( graph, principal=principal, registry=registry, ) key = payload.idempotency_key.strip() existing = session.scalar( select(WorkflowInstance).where( WorkflowInstance.tenant_id == tenant_id, WorkflowInstance.definition_id == definition.id, WorkflowInstance.idempotency_key == key, ) ) if existing is not None: if ( dict(existing.input_) != dict(payload.input) or existing.correlation_id != payload.correlation_id or existing.start_origin != normalized_origin ): raise WorkflowConflictError( "The Workflow idempotency key was already used with different input." ) return get_instance( session, tenant_id=tenant_id, instance_id=existing.id, ), True start_node = _start_node( graph, kind=_start_kind_for_origin(normalized_origin), ) now = utcnow() instance = WorkflowInstance( tenant_id=tenant_id, definition_id=definition.id, definition_revision_id=revision.id, status="running", start_origin=normalized_origin, idempotency_key=key, correlation_id=payload.correlation_id, input_=dict(payload.input), context_={"input": dict(payload.input), "steps": {}}, output_={}, authorization_=_authorization_payload( principal, graph=graph, registry=registry, ), started_at=now, created_by=actor_id, ) session.add(instance) session.flush() instance.authorization_ = { **dict(instance.authorization_), "authorization_ref": f"workflow-instance:{instance.id}", } _record_event( session, instance, kind="workflow.instance.started", actor_id=actor_id, payload={ "definition_ref": f"workflow-definition:{definition.id}", "revision": revision.revision, "definition_hash": revision.content_hash, "execution_mode": revision.execution_mode, "start_origin": normalized_origin, "input": dict(payload.input), }, ) _drive_instance( session, instance=instance, graph=graph, next_node_id=start_node.id, principal=principal, registry=registry, actor_id=actor_id, ) session.flush() return instance, False def reconcile_instance( session: Session, *, instance: WorkflowInstance, principal: ApiPrincipal, registry: object | None, actor_id: str | None = None, ) -> bool: if instance.status not in {"running", "waiting"}: return False step = _current_step(session, instance) if step is None: return False revision = session.get( WorkflowDefinitionRevision, instance.definition_revision_id, ) if revision is None: _fail_instance( session, instance, message="Pinned Workflow revision no longer exists.", ) return True graph = _runtime_graph(revision) node = _node(graph, step.node_id) if step.node_type == "workflow.capability": if str(step.handoff.get("state") or "") not in { "", "pending", "running", "retrying", }: return False return _execute_capability_step( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, ) if step.node_type != "workflow.dataflow": return False if not step.external_ref: _start_dataflow_step( session, instance=instance, step=step, node=node, principal=principal, registry=registry, graph=graph, actor_id=actor_id, ) return True provider = dataflow_run_lifecycle(registry) if provider is None: _set_dependency_handoff( session, instance=instance, step=step, message="The Dataflow module is not available.", ) return False try: descriptor = provider.get_run( session, principal, run_ref=step.external_ref, ) except ValueError as exc: _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message=str(exc), ) return True if descriptor is None: _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message="The linked Dataflow run no longer exists.", ) return True descriptor_recovery = descriptor.metadata.get("recovery") recovery_requires_attention = bool( isinstance(descriptor_recovery, Mapping) and descriptor_recovery.get("requires_attention") ) if descriptor.status == "outcome_unknown" or recovery_requires_attention: previous_state = str(step.handoff.get("state") or "") step.status = "waiting" step.error = ( "The Dataflow output outcome is unresolved. Reconcile the run " "before this Workflow can continue." ) step.handoff = { "kind": "dataflow_recovery", "state": "outcome_unknown", "run_ref": descriptor.ref, "pipeline_ref": descriptor.pipeline_ref, "action_url": _dataflow_action_url( descriptor.pipeline_ref, descriptor.ref, ), "allowed_actions": ["cancel"], "outcome_unknown": True, "recovery": ( dict(descriptor_recovery) if isinstance(descriptor_recovery, Mapping) else {"requires_attention": True} ), } instance.status = "waiting" instance.error = step.error if previous_state != "outcome_unknown": _record_event( session, instance, step=step, kind="workflow.dataflow.outcome_unknown", actor_id=actor_id, payload=dict(step.handoff), ) _notify_handoff( session, registry=registry, instance=instance, step=step, subject="Workflow Dataflow outcome requires reconciliation", ) return False if descriptor.status in {"queued", "retrying", "running"}: step.handoff = { **dict(step.handoff), "state": descriptor.status, "progress_percent": int(descriptor.metadata.get("progress_percent") or 0), "progress_phase": str( descriptor.metadata.get("progress_phase") or descriptor.status ), } return False if descriptor.status == "succeeded": return _handle_dataflow_success( session, instance=instance, step=step, node=node, graph=graph, descriptor=descriptor, principal=principal, registry=registry, actor_id=actor_id, ) if descriptor.status == "cancelled": _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message="The linked Dataflow run was cancelled.", state="cancelled", ) return True _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message=descriptor.error or "The linked Dataflow run failed.", ) return True def resolve_step( session: Session, *, tenant_id: str, instance_id: str, step_id: str, actor_id: str | None, principal: ApiPrincipal, registry: object | None, payload: WorkflowStepActionRequest, ) -> WorkflowInstance: instance = get_instance( session, tenant_id=tenant_id, instance_id=instance_id, for_update=True, ) if instance.status != "waiting" or instance.current_step_id != step_id: raise WorkflowConflictError( "Only the current waiting Workflow step can be resolved." ) step = session.get(WorkflowInstanceStep, step_id) if step is None or step.instance_id != instance.id: raise WorkflowNotFoundError("Workflow step not found.") revision = session.get( WorkflowDefinitionRevision, instance.definition_revision_id, ) if revision is None: raise WorkflowConflictError("Pinned Workflow revision no longer exists.") graph = _runtime_graph(revision) node = _node(graph, step.node_id) allowed_actions = { str(action) for action in step.handoff.get("allowed_actions") or () } if payload.action not in allowed_actions: raise WorkflowConflictError( f"Action {payload.action!r} is not available for this handoff." ) _record_event( session, instance, step=step, kind="workflow.step.action", actor_id=actor_id, payload={ "action": payload.action, "comment": payload.comment, "evidence": list(payload.evidence), "output": dict(payload.output), }, ) if payload.action == "cancel": return cancel_instance( session, tenant_id=tenant_id, instance_id=instance_id, actor_id=actor_id, principal=principal, registry=registry, ) if payload.action in {"confirm_effect", "confirm_absent"}: recovery_payload = step.handoff.get("recovery") if not isinstance(recovery_payload, Mapping): raise WorkflowConflictError( "This handoff has no durable recovery operation to reconcile." ) operation_id = str(recovery_payload.get("operation_id") or "").strip() if not operation_id: raise WorkflowConflictError( "This handoff has no durable recovery operation to reconcile." ) if not payload.evidence and not payload.comment: raise WorkflowConflictError( "Record an evidence reference or operator comment before " "resolving an unknown provider outcome." ) try: recovery = claim_workflow_action_recovery( session, operation_id=operation_id, ) except WorkflowRecoveryError as exc: raise WorkflowConflictError(str(exc)) from exc effect_occurred = payload.action == "confirm_effect" evidence = { "verified": True, "checks": { "operator_actor_ref": actor_id, "evidence_refs_sha256": canonical_sha256(list(payload.evidence)), "operator_output_sha256": canonical_sha256(dict(payload.output)), "operator_comment_sha256": canonical_sha256(payload.comment or ""), "effect_occurred": effect_occurred, }, } if not effect_occurred: previous_recovery = dict(recovery_payload) call_number = max(1, int(previous_recovery.get("call_number") or 1)) step.status = "waiting" step.error = None step.handoff = { **dict(step.handoff), "state": "retryable", "message": ( "The provider effect was verified absent. A deliberate retry " "is now safe." ), "allowed_actions": ["retry", "cancel"], "outcome_unknown": False, "recovery": { **previous_recovery, "status": "recovered", "requires_attention": False, "next_call_number": call_number + 1, }, } instance.status = "waiting" instance.error = None _record_event( session, instance, step=step, kind="workflow.action.effect_absent", actor_id=actor_id, payload={ "recovery_operation_id": operation_id, "evidence": list(payload.evidence), }, ) recovery.commit_unknown_resolution( session, effect_occurred=False, evidence=evidence, summary="Operator verified that the provider effect is absent", ) return instance output = { **dict(step.output_), **dict(payload.output), "recovery_decision": "effect_confirmed", "recovery_evidence": list(payload.evidence), "recovery_comment": payload.comment, } next_node_id = _complete_step( session, instance=instance, step=step, graph=graph, port="success", output=output, actor_id=actor_id, ) recovery.commit_unknown_resolution( session, effect_occurred=True, evidence=evidence, summary="Operator verified that the provider effect occurred", ) _drive_instance( session, instance=instance, graph=graph, next_node_id=next_node_id, principal=principal, registry=registry, actor_id=actor_id, ) return instance if payload.action == "changes": step.handoff = { **dict(step.handoff), "state": "changes_requested", "last_comment": payload.comment, } return instance if payload.action == "retry": if step.node_type == "workflow.capability": step.status = "running" step.error = None step.handoff = { **dict(step.handoff), "state": "retrying", } instance.status = "running" instance.error = None _execute_capability_step( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, ) return instance if step.node_type != "workflow.dataflow": raise WorkflowConflictError( "Only failed module-action or Dataflow handoffs can be retried." ) step.status = "superseded" step.finished_at = utcnow() step.completed_by = actor_id instance.status = "running" instance.current_step_id = None instance.error = None _drive_instance( session, instance=instance, graph=graph, next_node_id=node.id, principal=principal, registry=registry, actor_id=actor_id, ) return instance port = _action_port(step, payload.action) output = { **dict(step.output_), **dict(payload.output), "decision": payload.action, "comment": payload.comment, "evidence": list(payload.evidence), } if step.node_type == "workflow.wait": from govoplan_workflow_engine.backend.triggers import resolve_wait_state resolve_wait_state(session, step_id=step.id, status="resumed") next_node_id = _complete_step( session, instance=instance, step=step, graph=graph, port=port, output=output, actor_id=actor_id, ) _drive_instance( session, instance=instance, graph=graph, next_node_id=next_node_id, principal=principal, registry=registry, actor_id=actor_id, ) return instance def cancel_instance( session: Session, *, tenant_id: str, instance_id: str, actor_id: str | None, principal: ApiPrincipal, registry: object | None, ) -> WorkflowInstance: instance = get_instance( session, tenant_id=tenant_id, instance_id=instance_id, for_update=True, ) if instance.status in {"completed", "failed", "cancelled"}: raise WorkflowConflictError(f"Workflow instance is already {instance.status}.") now = utcnow() instance.cancellation_requested_at = now step = _current_step(session, instance) if step is not None: if step.external_ref: provider = dataflow_run_lifecycle(registry) if provider is not None: try: provider.cancel_run( session, principal, run_ref=step.external_ref, ) except ValueError as exc: logger.info( "Linked Dataflow run could not be cancelled for " "Workflow instance %s: %s", instance.id, exc, ) if step.node_type == "workflow.wait": from govoplan_workflow_engine.backend.triggers import ( resolve_wait_state, ) resolve_wait_state(session, step_id=step.id, status="cancelled") step.status = "cancelled" step.finished_at = now step.completed_by = actor_id instance.status = "cancelled" instance.finished_at = now instance.current_step_id = None instance.error = "Cancelled by request." _record_event( session, instance, step=step, kind="workflow.instance.cancelled", actor_id=actor_id, payload={"external_ref": step.external_ref if step else None}, ) return instance def reconcile_pending_instances( session: Session, *, registry: object | None, tenant_id: str | None = None, limit: int = 50, ) -> dict[str, object]: clauses = [ WorkflowInstance.status.in_(("running", "waiting")), WorkflowInstance.current_step_id.is_not(None), ] if tenant_id: clauses.append(WorkflowInstance.tenant_id == tenant_id) instances = list( session.scalars( select(WorkflowInstance) .where(*clauses) .order_by(WorkflowInstance.updated_at, WorkflowInstance.id) .limit(max(1, min(int(limit), 200))) .with_for_update(skip_locked=True) ) ) summary: dict[str, object] = { "inspected": len(instances), "advanced": 0, "waiting": 0, "failed": 0, "skipped": 0, } for instance in instances: try: fence = acquire_workflow_state_fence( session, resource_key=f"workflow:instance:{instance.id}", ) except RuntimeError: logger.warning( "Workflow instance %s could not acquire a runtime fence", instance.id, exc_info=True, ) summary["skipped"] = int(summary["skipped"]) + 1 continue if fence is None: summary["skipped"] = int(summary["skipped"]) + 1 continue step = _current_step(session, instance) if step is None or step.node_type not in { "workflow.capability", "workflow.dataflow", }: summary["waiting"] = int(summary["waiting"]) + 1 release_workflow_state_fence(session, fence) continue principal = _resolve_instance_principal( session, instance=instance, registry=registry, ) if principal is None: summary["skipped"] = int(summary["skipped"]) + 1 release_workflow_state_fence(session, fence) continue changed = reconcile_instance( session, instance=instance, principal=principal, registry=registry, ) if instance.status == "failed": summary["failed"] = int(summary["failed"]) + 1 elif changed: summary["advanced"] = int(summary["advanced"]) + 1 else: summary["waiting"] = int(summary["waiting"]) + 1 release_workflow_state_fence(session, fence) session.flush() return summary def instance_response( session: Session, instance: WorkflowInstance, *, replayed: bool = False, ) -> WorkflowInstanceResponse: revision = session.get( WorkflowDefinitionRevision, instance.definition_revision_id, ) definition = session.get(WorkflowDefinition, instance.definition_id) if revision is None or definition is None: raise WorkflowNotFoundError( "Workflow instance definition evidence is incomplete." ) view_context = _instance_view_context(instance, revision) return WorkflowInstanceResponse( id=instance.id, definition_id=instance.definition_id, definition_name=definition.name, definition_revision=revision.revision, definition_hash=revision.content_hash, execution_mode=revision.execution_mode, # type: ignore[arg-type] start_origin=instance.start_origin, # type: ignore[arg-type] view_context=view_context, status=instance.status, # type: ignore[arg-type] idempotency_key=instance.idempotency_key, correlation_id=instance.correlation_id, current_step_id=instance.current_step_id, input=dict(instance.input_), context=dict(instance.context_), output=dict(instance.output_), started_at=instance.started_at, finished_at=instance.finished_at, cancellation_requested_at=instance.cancellation_requested_at, error=instance.error, created_by=instance.created_by, created_at=instance.created_at, updated_at=instance.updated_at, steps=[ WorkflowInstanceStepResponse( id=step.id, sequence=step.sequence, node_id=step.node_id, node_type=step.node_type, status=step.status, # type: ignore[arg-type] attempt=step.attempt, input=dict(step.input_), output=dict(step.output_), handoff=dict(step.handoff), external_ref=step.external_ref, started_at=step.started_at, finished_at=step.finished_at, error=step.error, completed_by=step.completed_by, created_at=step.created_at, updated_at=step.updated_at, ) for step in instance.steps ], events=[ WorkflowInstanceEventResponse( id=event.id, sequence=event.sequence, step_id=event.step_id, kind=event.kind, actor_id=event.actor_id, payload=dict(event.payload), created_at=event.created_at, ) for event in instance.events ], replayed=replayed, ) def _drive_instance( session: Session, *, instance: WorkflowInstance, graph: WorkflowGraph, next_node_id: str | None, principal: ApiPrincipal, registry: object | None, actor_id: str | None, ) -> None: transitions = 0 while ( next_node_id is not None and instance.status == "running" and transitions < MAX_INSTANCE_TRANSITIONS ): transitions += 1 node = _node(graph, next_node_id) step = _new_step( session, instance=instance, node=node, ) instance.current_step_id = step.id _record_event( session, instance, step=step, kind="workflow.step.started", actor_id=actor_id, payload={"node_id": node.id, "node_type": node.type}, ) if node.type.startswith("workflow.start."): next_node_id = _complete_step( session, instance=instance, step=step, graph=graph, port="output", output={"input": dict(instance.input_)}, actor_id=actor_id, ) continue if node.type == "workflow.dataflow": _start_dataflow_step( session, instance=instance, step=step, node=node, principal=principal, registry=registry, graph=graph, actor_id=actor_id, ) return if node.type == "workflow.capability": _execute_capability_step( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, ) return if node.type in { "workflow.activity", "workflow.review", }: _set_human_handoff( session, instance=instance, step=step, node=node, registry=registry, ) return if node.type == "workflow.wait": from govoplan_workflow_engine.backend.triggers import ( register_wait_state, ) wait_state = register_wait_state( session, instance=instance, step=step, node=node, ) if wait_state is None: _set_human_handoff( session, instance=instance, step=step, node=node, registry=registry, ) else: _set_automated_wait( session, instance=instance, step=step, node=node, wait_state=wait_state, ) return if node.type == "workflow.end.completed": _complete_step( session, instance=instance, step=step, graph=graph, port="output", output=dict(instance.context_), actor_id=actor_id, ) instance.status = "completed" instance.finished_at = utcnow() instance.current_step_id = None instance.output_ = dict(instance.context_) _record_event( session, instance, step=step, kind="workflow.instance.completed", actor_id=actor_id, payload={"output": dict(instance.output_)}, ) return if node.type == "workflow.end.cancelled": step.status = "completed" step.finished_at = utcnow() instance.status = "cancelled" instance.finished_at = utcnow() instance.current_step_id = None _record_event( session, instance, step=step, kind="workflow.instance.cancelled", actor_id=actor_id, payload={"reason": node.config.get("reason")}, ) return _set_dependency_handoff( session, instance=instance, step=step, message=( f"Runtime support for {node.type} requires an explicit " "operator transition." ), ) return if next_node_id is None and instance.status == "running": _fail_instance( session, instance, message="Workflow reached a step without a configured transition.", ) return if transitions >= MAX_INSTANCE_TRANSITIONS: _fail_instance( session, instance, message="Workflow exceeded the bounded transition limit.", ) def _execute_capability_step( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, node: WorkflowNode, graph: WorkflowGraph, principal: ApiPrincipal, registry: object | None, actor_id: str | None, ) -> bool: try: provider, definition = _capability_action_definition( node, principal=principal, registry=registry, ) execution_context = { **dict(instance.context_), "instance": { "id": instance.id, "definition_id": instance.definition_id, "correlation_id": instance.correlation_id, }, "step": { "id": step.id, "node_id": step.node_id, "sequence": step.sequence, "attempt": step.attempt, }, } action_input = _mapped_action_input( node.config.get("input_mapping"), execution_context, ) request = ActionExecutionRequest( tenant_id=instance.tenant_id, action_key=definition.action_key, input=action_input, idempotency_key=_action_idempotency_key( node, step=step, capability_name=str(node.config.get("capability") or ""), action_key=definition.action_key, context=execution_context, ), invocation=AutomationInvocation( kind="workflow", trigger_ref=f"workflow-instance:{instance.id}", correlation_id=instance.correlation_id, causation_id=f"workflow-step:{step.id}", requested_by=instance.created_by, metadata={ "workflow_definition_ref": ( f"workflow-definition:{instance.definition_id}" ), "workflow_node_id": node.id, }, ), actor_ref=instance.created_by, metadata={ "workflow_instance_ref": f"workflow-instance:{instance.id}", "workflow_step_ref": f"workflow-step:{step.id}", }, ) preview = provider.preview_action( session, principal, request=request, ) except WorkflowConflictError as exc: _set_action_handoff( session, instance=instance, step=step, state="blocked", message=str(exc), action_key=str(node.config.get("operation") or ""), capability_name=str(node.config.get("capability") or ""), registry=registry, ) return True except ValueError as exc: _set_action_handoff( session, instance=instance, step=step, state="blocked", message=str(exc), action_key=str(node.config.get("operation") or ""), capability_name=str(node.config.get("capability") or ""), registry=registry, ) return True except Exception as exc: logger.exception( "Workflow action preview failed for instance %s step %s", instance.id, step.id, ) _set_action_handoff( session, instance=instance, step=step, state="quarantined", message=( "The module action preview failed unexpectedly. No action " "execution was attempted." ), action_key=str(node.config.get("operation") or ""), capability_name=str(node.config.get("capability") or ""), registry=registry, details={"error_type": type(exc).__name__}, ) return True if not isinstance(preview, ActionPreview): _set_action_handoff( session, instance=instance, step=step, state="quarantined", message="The action provider returned an invalid preview.", action_key=definition.action_key, capability_name=str(node.config.get("capability") or ""), registry=registry, ) return True preview_payload = _action_preview_payload(preview) if preview.action_key != definition.action_key: _set_action_handoff( session, instance=instance, step=step, state="quarantined", message="The action provider returned a preview for another action.", action_key=definition.action_key, capability_name=str(node.config.get("capability") or ""), registry=registry, details={"preview": preview_payload}, ) return True if not preview.allowed: _set_action_handoff( session, instance=instance, step=step, state="blocked", message=preview.summary or "The module action is not allowed.", action_key=definition.action_key, capability_name=str(node.config.get("capability") or ""), registry=registry, details={"preview": preview_payload}, ) return True request = ActionExecutionRequest( tenant_id=request.tenant_id, action_key=request.action_key, input=request.input, idempotency_key=request.idempotency_key, invocation=request.invocation, actor_ref=request.actor_ref, preview_ref=preview.preview_ref, metadata=request.metadata, ) capability_name = str(node.config.get("capability") or "") revision = session.get( WorkflowDefinitionRevision, instance.definition_revision_id, ) if revision is None: _set_action_handoff( session, instance=instance, step=step, state="blocked", message="The pinned Workflow revision is unavailable.", action_key=definition.action_key, capability_name=capability_name, registry=registry, ) return True try: action_recovery = begin_workflow_action_recovery( session, instance=instance, step=step, revision=revision, definition=definition, capability_name=capability_name, request_idempotency_key=request.idempotency_key, action_input=action_input, preview_payload=preview_payload, backup_reference=( str(node.config.get("recovery_backup_reference") or "").strip() or None ), approval_reference=( str(node.config.get("recovery_approval_reference") or "").strip() or None ), ) except WorkflowRecoveryBusy as exc: _set_action_handoff( session, instance=instance, step=step, state="running", message=str(exc), action_key=definition.action_key, capability_name=capability_name, registry=registry, ) return False except WorkflowRecoveryConflict as exc: recovery_status = exc.status if recovery_status == "running": try: recovery_status = reconcile_stale_workflow_action_recovery( session, operation_id=exc.operation_id, ) except WorkflowRecoveryBusy: recovery_status = "running" outcome_unknown = recovery_status in { "outcome_unknown", "recovery_required", "manual_intervention", } safe_to_retry = recovery_status in {"failed", "recovered", "rejected"} previous_recovery = ( dict(step.handoff.get("recovery")) if isinstance(step.handoff.get("recovery"), Mapping) else {} ) call_number = max(1, int(previous_recovery.get("call_number") or 1)) _set_action_handoff( session, instance=instance, step=step, state=( "outcome_unknown" if outcome_unknown else "retryable" if safe_to_retry else recovery_status ), message=( "The provider outcome must be reconciled before this Workflow " "can continue." if outcome_unknown else ( "The stale action was proven not to have committed. A " "deliberate retry is now safe." if safe_to_retry else str(exc) ) ), action_key=definition.action_key, capability_name=capability_name, registry=registry, details={ "outcome_unknown": outcome_unknown, "recovery": { **previous_recovery, "operation_id": exc.operation_id, "status": recovery_status, "requires_attention": outcome_unknown, **( {"next_call_number": call_number + 1} if safe_to_retry else {} ), }, }, ) return True except WorkflowRecoveryError as exc: _set_action_handoff( session, instance=instance, step=step, state="blocked", message=str(exc), action_key=definition.action_key, capability_name=capability_name, registry=registry, details={"recovery_unavailable": True}, ) return True if action_recovery.replayed: session.expire_all() return True action_recovery.checkpoint_dispatch( session, instance=instance, step=step, action_key=definition.action_key, capability_name=capability_name, ) try: result = provider.execute_action( session, principal, request=request, ) except Exception as exc: logger.exception( "Workflow action execution failed for instance %s step %s", instance.id, step.id, ) session.rollback() if action_recovery.mode.value == "atomic": _set_action_handoff( session, instance=instance, step=step, state="retryable", message=( "The atomic module action failed and its database changes " "were rolled back." ), action_key=definition.action_key, capability_name=capability_name, registry=registry, details={ "preview": preview_payload, "error_type": type(exc).__name__, "recovery": { "operation_id": action_recovery.operation_id, "mode": action_recovery.mode.value, "status": "failed", "call_number": action_recovery.call_number, "next_call_number": action_recovery.call_number + 1, "requires_attention": False, }, }, ) action_recovery.commit_definitive_failure( session, summary="The atomic provider call failed before commit", error_type=type(exc).__name__, ) return True _set_action_handoff( session, instance=instance, step=step, state="outcome_unknown", message=( "The module action outcome is unknown. Inspect the provider " "and record evidence before continuing or retrying." ), action_key=definition.action_key, capability_name=capability_name, registry=registry, details={ "preview": preview_payload, "error_type": type(exc).__name__, "outcome_unknown": True, "recovery": { "operation_id": action_recovery.operation_id, "mode": action_recovery.mode.value, "status": "outcome_unknown", "call_number": action_recovery.call_number, "requires_attention": True, }, }, ) action_recovery.commit_unknown( session, error_type=type(exc).__name__, message=( "Inspect the provider by stable idempotency key before any retry" ), ) return True if not isinstance(result, ActionExecutionResult): return _handle_invalid_action_result( session, instance=instance, step=step, message="The action provider returned an invalid execution result.", definition=definition, capability_name=capability_name, registry=registry, preview_payload=preview_payload, action_recovery=action_recovery, error_type="InvalidActionExecutionResult", ) allowed_states = { "pending", "running", "completed", "blocked", "retryable", "quarantined", "manual_required", "compensation_required", } announced_effects = {item.effect_key for item in provider.effect_definitions()} unknown_effects = sorted( { effect.effect_key for effect in result.observed_effects if effect.effect_key not in announced_effects } ) if result.state not in allowed_states or unknown_effects: return _handle_invalid_action_result( session, instance=instance, step=step, message=( "The action provider returned an unsupported state." if result.state not in allowed_states else "The action provider reported unannounced effects." ), definition=definition, capability_name=capability_name, registry=registry, preview_payload=preview_payload, action_recovery=action_recovery, error_type=( "UnsupportedActionState" if result.state not in allowed_states else "UnannouncedActionEffect" ), extra_details={ "state": result.state, "unknown_effects": unknown_effects, }, ) result_payload = _action_result_payload(result) step.output_ = { "action_key": definition.action_key, "capability": capability_name, "idempotency_key": request.idempotency_key, "preview": preview_payload, "execution": result_payload, } if result.state != "completed": _set_action_handoff( session, instance=instance, step=step, state=result.state, message=( result.error or result.manual_instructions or f"Module action is {result.state.replace('_', ' ')}." ), action_key=definition.action_key, capability_name=capability_name, registry=registry, details={ "preview": preview_payload, "execution": result_payload, "recovery": { "operation_id": action_recovery.operation_id, "mode": action_recovery.mode.value, "status": "succeeded", "call_number": action_recovery.call_number, "next_call_number": action_recovery.call_number + 1, "requires_attention": False, }, }, ) action_recovery.commit_conclusive_result( session, provider_state=result.state, result_sha256=canonical_sha256(result_payload), observed_effects_sha256=canonical_sha256( result_payload["observed_effects"] ), ) return True _record_event( session, instance, step=step, kind="workflow.action.completed", actor_id=actor_id, payload={ "action_key": definition.action_key, "capability": capability_name, "idempotency_key": request.idempotency_key, "observed_effects": result_payload["observed_effects"], "audit_event_refs": result_payload["audit_event_refs"], }, ) suggested_port = str(result.output.get("outcome") or "success") if suggested_port not in {"success", "warning"}: suggested_port = "success" next_node_id = _complete_step( session, instance=instance, step=step, graph=graph, port=suggested_port, output=dict(step.output_), actor_id=actor_id, ) action_recovery.commit_conclusive_result( session, provider_state=result.state, result_sha256=canonical_sha256(result_payload), observed_effects_sha256=canonical_sha256( result_payload["observed_effects"] ), ) _drive_instance( session, instance=instance, graph=graph, next_node_id=next_node_id, principal=principal, registry=registry, actor_id=actor_id, ) return True def _capability_action_definition( node: WorkflowNode, *, principal: ApiPrincipal, registry: object | None, ) -> tuple[object, ActionDefinition]: capability_name = str(node.config.get("capability") or "").strip() action_key = str(node.config.get("operation") or "").strip() if not capability_name or not action_key: raise WorkflowConflictError( "Module-action steps require a capability and operation." ) provider = action_effect_provider(registry, capability_name) if provider is None: raise WorkflowConflictError( f"Action capability {capability_name!r} is not available." ) definitions = [ item for item in provider.action_definitions() if item.action_key == action_key ] if len(definitions) != 1: raise WorkflowConflictError( f"Action {action_key!r} is not uniquely announced by {capability_name!r}." ) definition = definitions[0] missing_scopes = [ scope for scope in definition.required_scopes if not has_scope(principal, scope) ] if missing_scopes: raise WorkflowConflictError( "Module action requires scopes: " + ", ".join(sorted(missing_scopes)) ) missing_capabilities = [ capability for capability in definition.required_capabilities if ( registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(capability) ) ] if missing_capabilities: raise WorkflowConflictError( "Module action requires capabilities: " + ", ".join(sorted(missing_capabilities)) ) effect_keys = {item.effect_key for item in provider.effect_definitions()} missing_effects = sorted(set(definition.expected_effect_keys) - effect_keys) if missing_effects: raise WorkflowConflictError( "Action provider does not define its expected effects: " + ", ".join(missing_effects) ) return provider, definition def _mapped_action_input( raw_mapping: object, context: Mapping[str, object], ) -> dict[str, object]: if raw_mapping is None or raw_mapping == "": return dict(context) if not isinstance(raw_mapping, Mapping): raise WorkflowConflictError("Module-action input mapping must be an object.") return { str(key): _resolve_action_value(value, context, depth=0) for key, value in raw_mapping.items() if str(key).strip() } def _resolve_action_value( value: object, context: Mapping[str, object], *, depth: int, ) -> object: if depth > 10: raise WorkflowConflictError("Module-action input mapping is nested too deeply.") if isinstance(value, str) and value.startswith("$"): path = value[1:].lstrip(".") current: object = context if not path: return dict(context) for segment in path.split("."): if not isinstance(current, Mapping) or segment not in current: raise WorkflowConflictError( f"Module-action input path {value!r} is unavailable." ) current = current[segment] return current if isinstance(value, Mapping): return { str(key): _resolve_action_value( nested, context, depth=depth + 1, ) for key, nested in value.items() } if isinstance(value, list): return [_resolve_action_value(item, context, depth=depth + 1) for item in value] return value def _action_idempotency_key( node: WorkflowNode, *, step: WorkflowInstanceStep, capability_name: str, action_key: str, context: Mapping[str, object], ) -> str: expression = str(node.config.get("idempotency_key") or "workflow-step").strip() if expression == "workflow-step": return step.idempotency_key resolved = _resolve_action_value(expression, context, depth=0) key = f"{capability_name}:{action_key}:{resolved}" if len(key) <= 255: return key digest = hashlib.sha256(key.encode("utf-8")).hexdigest() return f"{capability_name[:80]}:{action_key[:80]}:{digest}" def _action_preview_payload(preview: object) -> dict[str, object]: return { "action_key": str(getattr(preview, "action_key", "")), "allowed": bool(getattr(preview, "allowed", False)), "summary": str(getattr(preview, "summary", "")), "risk_level": str(getattr(preview, "risk_level", "")), "reversibility": str(getattr(preview, "reversibility", "")), "preview_ref": getattr(preview, "preview_ref", None), "blockers": list(getattr(preview, "blockers", ()) or ()), "policy_provenance": [ dict(item) for item in getattr(preview, "policy_provenance", ()) or () ], "effects": [ { "effect_key": item.effect_key, "summary": item.summary, "resource_refs": list(item.resource_refs), "external_system_refs": list(item.external_system_refs), } for item in getattr(preview, "effects", ()) or () ], } def _action_result_payload( result: ActionExecutionResult, ) -> dict[str, object]: return { "state": result.state, "output": dict(result.output), "observed_effects": [ { "effect_key": effect.effect_key, "operation": effect.operation, "resource_ref": effect.resource_ref, "external_system_ref": effect.external_system_ref, "audit_event_ref": effect.audit_event_ref, "summary": effect.summary, "metadata": dict(effect.metadata), } for effect in result.observed_effects ], "error": result.error, "retry_after": ( result.retry_after.isoformat() if result.retry_after is not None else None ), "manual_instructions": result.manual_instructions, "compensation_action_key": result.compensation_action_key, "audit_event_refs": list(result.audit_event_refs), } def _handle_invalid_action_result( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, message: str, definition: ActionDefinition, capability_name: str, registry: object | None, preview_payload: Mapping[str, object], action_recovery: WorkflowActionRecovery, error_type: str, extra_details: Mapping[str, object] | None = None, ) -> bool: session.rollback() recovery_details = { "operation_id": action_recovery.operation_id, "mode": action_recovery.mode.value, "call_number": action_recovery.call_number, } if action_recovery.mode.value == "atomic": _set_action_handoff( session, instance=instance, step=step, state="retryable", message=f"{message} Atomic database changes were rolled back.", action_key=definition.action_key, capability_name=capability_name, registry=registry, details={ "preview": dict(preview_payload), "error_type": error_type, **dict(extra_details or {}), "recovery": { **recovery_details, "status": "failed", "next_call_number": action_recovery.call_number + 1, "requires_attention": False, }, }, ) action_recovery.commit_definitive_failure( session, summary=message, error_type=error_type, ) return True _set_action_handoff( session, instance=instance, step=step, state="outcome_unknown", message=f"{message} Reconcile the provider before retrying.", action_key=definition.action_key, capability_name=capability_name, registry=registry, details={ "preview": dict(preview_payload), "error_type": error_type, "outcome_unknown": True, **dict(extra_details or {}), "recovery": { **recovery_details, "status": "outcome_unknown", "requires_attention": True, }, }, ) action_recovery.commit_unknown( session, error_type=error_type, message="Inspect the provider by stable idempotency key before any retry", ) return True def _set_action_handoff( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, state: str, message: str, action_key: str, capability_name: str, registry: object | None, details: Mapping[str, object] | None = None, ) -> None: previous = dict(step.handoff) if state in {"pending", "running"}: allowed_actions = ["cancel"] elif state in {"outcome_unknown", "recovery_required"}: allowed_actions = ["confirm_effect", "confirm_absent", "cancel"] elif state == "compensation_required": allowed_actions = ["reject", "cancel"] else: allowed_actions = ["retry", "reject", "cancel"] details_payload = dict(details or {}) if "recovery" not in details_payload and isinstance( previous.get("recovery"), Mapping ): details_payload["recovery"] = dict(previous["recovery"]) step.status = "waiting" step.error = message if state not in {"pending", "running"} else None step.handoff = { "kind": "module_action", "state": state, "message": message, "action_key": action_key, "capability": capability_name, "allowed_actions": allowed_actions, "suggested_port": "failure", **details_payload, } instance.status = "waiting" instance.error = step.error if previous.get("state") != state or previous.get("message") != message: _record_event( session, instance, step=step, kind=f"workflow.action.{state}", actor_id=None, payload=dict(step.handoff), ) if state not in {"pending", "running"}: _notify_handoff( session, registry=registry, instance=instance, step=step, subject=f"Workflow action requires attention: {action_key}", ) def _start_dataflow_step( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, node: WorkflowNode, principal: ApiPrincipal, registry: object | None, graph: WorkflowGraph, actor_id: str | None, ) -> None: provider = dataflow_run_lifecycle(registry) if provider is None: _set_dependency_handoff( session, instance=instance, step=step, message="Enable Dataflow to execute this Workflow step.", ) return pipeline_ref = str(node.config.get("pipeline_ref") or "").strip() try: revision = int(node.config.get("revision") or 0) row_limit = max( 1, min(int(node.config.get("row_limit") or 500), 10_000), ) except (TypeError, ValueError): _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message="Dataflow revision and row limit must be integers.", ) return if not pipeline_ref or revision < 1: _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message="Dataflow steps require a pipeline and pinned revision.", ) return target_ref = str(node.config.get("publication_target_ref") or "").strip() try: run = provider.start_run( session, principal, request=DataflowRunRequest( pipeline_ref=pipeline_ref, revision=revision, idempotency_key=step.idempotency_key, row_limit=row_limit, environment=str(node.config.get("environment") or "development"), publication=( DataflowPublicationTarget(target_datasource_ref=target_ref) if target_ref else None ), invocation=AutomationInvocation( kind="workflow", correlation_id=instance.correlation_id, causation_id=f"workflow-step:{step.id}", requested_by=instance.created_by, metadata={ "workflow_instance_ref": (f"workflow-instance:{instance.id}"), "workflow_step_ref": f"workflow-step:{step.id}", }, ), ), ) except ValueError as exc: _handle_dataflow_failure( session, instance=instance, step=step, node=node, graph=graph, principal=principal, registry=registry, actor_id=actor_id, message=str(exc), ) return step.external_ref = run.ref step.status = "waiting" step.output_ = _dataflow_output(run) step.handoff = { "kind": "dataflow_run", "state": run.status, "run_ref": run.ref, "pipeline_ref": pipeline_ref, "pipeline_revision": revision, "action_url": _dataflow_action_url(pipeline_ref, run.ref), "allowed_actions": ["cancel"], "progress_percent": int(run.metadata.get("progress_percent") or 0), "progress_phase": str(run.metadata.get("progress_phase") or run.status), } instance.status = "waiting" _record_event( session, instance, step=step, kind="workflow.dataflow.started", actor_id=instance.created_by, payload=dict(step.handoff), ) def _handle_dataflow_success( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, node: WorkflowNode, graph: WorkflowGraph, descriptor: DataflowRunDescriptor, principal: ApiPrincipal, registry: object | None, actor_id: str | None, ) -> bool: diagnostics = descriptor.metadata.get("diagnostics") items = diagnostics if isinstance(diagnostics, list) else [] warnings = [ dict(item) for item in items if isinstance(item, Mapping) and str(item.get("severity") or "") == "warning" ] explicit_review = any( str(item.get("code") or "") in { "review.required", "reconciliation.review_required", } for item in items if isinstance(item, Mapping) ) output = _dataflow_output(descriptor) step.output_ = output if explicit_review or ( warnings and str(node.config.get("warning_policy") or "review") == "review" ): step.status = "waiting" step.handoff = { "kind": "dataflow_review", "state": "review_required", "run_ref": descriptor.ref, "pipeline_ref": descriptor.pipeline_ref, "action_url": _dataflow_action_url( descriptor.pipeline_ref, descriptor.ref, ), "allowed_actions": [ "approve", "changes", "reject", "retry", "cancel", ], "suggested_port": ("review_required" if explicit_review else "warning"), "warnings": warnings, "output": output, } instance.status = "waiting" _record_event( session, instance, step=step, kind="workflow.dataflow.review_required", actor_id=actor_id, payload=dict(step.handoff), ) _notify_handoff( session, registry=registry, instance=instance, step=step, subject="Workflow Dataflow review required", ) return True port = "warning" if warnings else "success" if port == "warning" and _next_node_id(graph, node.id, port) is None: port = "success" next_node_id = _complete_step( session, instance=instance, step=step, graph=graph, port=port, output=output, actor_id=actor_id, ) _drive_instance( session, instance=instance, graph=graph, next_node_id=next_node_id, principal=principal, registry=registry, actor_id=actor_id, ) return True def _complete_step( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, graph: WorkflowGraph, port: str, output: Mapping[str, object], actor_id: str | None, ) -> str | None: step.status = "completed" step.output_ = dict(output) step.finished_at = utcnow() step.completed_by = actor_id step.handoff = {} context = dict(instance.context_) step_values = dict(context.get("steps") or {}) step_values[step.node_id] = dict(output) context["steps"] = step_values instance.context_ = context instance.status = "running" instance.current_step_id = None _record_event( session, instance, step=step, kind="workflow.step.completed", actor_id=actor_id, payload={"port": port, "output": dict(output)}, ) return _next_node_id(graph, step.node_id, port) def _set_human_handoff( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, node: WorkflowNode, registry: object | None, ) -> None: if node.type == "workflow.review": actions = ["approve", "changes", "reject", "cancel"] kind = "review" elif node.type == "workflow.wait": actions = ["resume", "cancel"] kind = "wait" else: actions = ["complete", "cancel"] kind = "activity" step.status = "waiting" step.handoff = { "kind": kind, "state": "waiting", "title": str(node.config.get("title") or node.label or node.type), "instructions": str(node.config.get("instructions") or ""), "assignee": node.config.get("reviewer") or node.config.get("assignee"), "required_evidence": list(node.config.get("required_evidence") or []), "allowed_actions": actions, } instance.status = "waiting" _record_event( session, instance, step=step, kind="workflow.handoff.created", actor_id=instance.created_by, payload=dict(step.handoff), ) _notify_handoff( session, registry=registry, instance=instance, step=step, subject=str(step.handoff["title"]), ) def _set_automated_wait( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, node: WorkflowNode, wait_state: object, ) -> None: mode = str(getattr(wait_state, "mode")) due_at = getattr(wait_state, "due_at") event_type = getattr(wait_state, "event_type") step.status = "waiting" step.handoff = { "kind": "event_wait" if mode == "event" else "timer", "state": "waiting", "title": str(node.config.get("title") or node.label or "Wait"), "mode": mode, "due_at": due_at.isoformat() if due_at else None, "event_type": event_type, "allowed_actions": ["cancel"], } instance.status = "waiting" _record_event( session, instance, step=step, kind="workflow.wait.registered", actor_id=instance.created_by, payload=dict(step.handoff), ) def _set_dependency_handoff( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, message: str, ) -> None: step.status = "waiting" step.error = message step.handoff = { "kind": "dependency", "state": "blocked", "message": message, "allowed_actions": ["retry", "cancel"], } instance.status = "waiting" instance.error = message _record_event( session, instance, step=step, kind="workflow.step.blocked", actor_id=None, payload=dict(step.handoff), ) def _handle_dataflow_failure( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, node: WorkflowNode, graph: WorkflowGraph, principal: ApiPrincipal, registry: object | None, actor_id: str | None, message: str, state: str = "failed", ) -> None: policy = str(node.config.get("failure_policy") or "manual") if policy == "manual": _set_failure_handoff( session, instance=instance, step=step, message=message, state=state, ) return step.error = message step.output_ = { **dict(step.output_), "status": state, "error": message, } if policy == "continue": _record_event( session, instance, step=step, kind="workflow.dataflow.failure_continued", actor_id=actor_id, payload={ "error": message, "external_ref": step.external_ref, }, ) next_node_id = _complete_step( session, instance=instance, step=step, graph=graph, port="failure", output=dict(step.output_), actor_id=actor_id, ) _drive_instance( session, instance=instance, graph=graph, next_node_id=next_node_id, principal=principal, registry=registry, actor_id=actor_id, ) return step.status = "failed" step.finished_at = utcnow() _record_event( session, instance, step=step, kind="workflow.dataflow.failed", actor_id=actor_id, payload={ "error": message, "external_ref": step.external_ref, "failure_policy": "fail", }, ) _fail_instance(session, instance, message=message) def _set_failure_handoff( session: Session, *, instance: WorkflowInstance, step: WorkflowInstanceStep, message: str, state: str = "failed", ) -> None: step.status = "waiting" step.error = message step.handoff = { "kind": "dataflow_failure", "state": state, "message": message, "run_ref": step.external_ref, "allowed_actions": ["retry", "reject", "cancel"], "suggested_port": "failure", } instance.status = "waiting" instance.error = message _record_event( session, instance, step=step, kind="workflow.dataflow.failed", actor_id=None, payload=dict(step.handoff), ) def _fail_instance( session: Session, instance: WorkflowInstance, *, message: str, ) -> None: instance.status = "failed" instance.finished_at = utcnow() instance.error = message instance.current_step_id = None _record_event( session, instance, kind="workflow.instance.failed", actor_id=None, payload={"error": message}, ) def _new_step( session: Session, *, instance: WorkflowInstance, node: WorkflowNode, ) -> WorkflowInstanceStep: sequence = ( int( session.scalar( select(func.max(WorkflowInstanceStep.sequence)).where( WorkflowInstanceStep.instance_id == instance.id ) ) or 0 ) + 1 ) attempt = ( int( session.scalar( select(func.count()) .select_from(WorkflowInstanceStep) .where( WorkflowInstanceStep.instance_id == instance.id, WorkflowInstanceStep.node_id == node.id, ) ) or 0 ) + 1 ) step = WorkflowInstanceStep( tenant_id=instance.tenant_id, instance=instance, sequence=sequence, node_id=node.id, node_type=node.type, status="running", attempt=attempt, idempotency_key=(f"workflow:{instance.id}:node:{node.id}:attempt:{attempt}"), input_=dict(instance.context_), output_={}, handoff={}, started_at=utcnow(), ) session.add(step) session.flush() return step def _record_event( session: Session, instance: WorkflowInstance, *, kind: str, actor_id: str | None, payload: Mapping[str, object], step: WorkflowInstanceStep | None = None, ) -> None: sequence = ( int( session.scalar( select(func.max(WorkflowInstanceEvent.sequence)).where( WorkflowInstanceEvent.instance_id == instance.id ) ) or 0 ) + 1 ) event = WorkflowInstanceEvent( tenant_id=instance.tenant_id, instance=instance, step_id=step.id if step else None, sequence=sequence, kind=kind, actor_id=actor_id, payload=dict(payload), created_at=utcnow(), ) session.add(event) session.flush() if kind.startswith("workflow.instance."): from govoplan_workflow_engine.backend.runtime import get_registry emit_platform_event( session, PlatformEvent( type=kind, module_id="workflow_engine", event_id=event.id, occurred_at=event.created_at, correlation_id=instance.correlation_id, causation_id=( str(payload.get("event_id")) if payload.get("event_id") else None ), actor=( EventActorRef(type="account", id=actor_id) if actor_id else EventActorRef(type="system_actor") ), tenant=EventTenantRef(id=instance.tenant_id), subject=EventObjectRef( type="workflow_instance", id=instance.id, ), resource=EventObjectRef( type="workflow_definition", id=instance.definition_id, ), classification="internal", payload={ "instance_id": instance.id, "definition_id": instance.definition_id, "definition_revision_id": instance.definition_revision_id, "step_id": step.id if step else None, "status": instance.status, "start_origin": instance.start_origin, }, ), registry=get_registry(), ) def _current_step( session: Session, instance: WorkflowInstance, ) -> WorkflowInstanceStep | None: if not instance.current_step_id: return None return session.get(WorkflowInstanceStep, instance.current_step_id) def _start_node(graph: WorkflowGraph, *, kind: str) -> WorkflowNode: expected = f"workflow.start.{kind}" node = next((item for item in graph.nodes if item.type == expected), None) if node is None: starts = [ item for item in graph.nodes if item.type.startswith("workflow.start.") ] if len(starts) == 1: node = starts[0] if node is None: raise WorkflowConflictError(f"Workflow has no {kind} start node.") return node def _normalize_start_origin(value: str) -> str: normalized = value.strip().lower() allowed = { "user", "api", "schedule", "event", "parent_workflow", "dependency", "retry", "replay", "backfill", } if normalized not in allowed: raise WorkflowConflictError(f"Unsupported Workflow start origin {value!r}.") return normalized def _start_kind_for_origin(origin: str) -> str: return { "user": "manual", "api": "api", "schedule": "schedule", "event": "event", "parent_workflow": "workflow", "dependency": "workflow", "retry": "api", "replay": "api", "backfill": "api", }[origin] def _instance_view_context( instance: WorkflowInstance, revision: WorkflowDefinitionRevision, ) -> WorkflowViewContextResponse | None: if not revision.view_id or instance.status not in {"running", "waiting"}: return None step = next( (item for item in instance.steps if item.id == instance.current_step_id), None, ) node = None if step is not None: graph = _runtime_graph(revision) node = next( (item for item in graph.nodes if item.id == step.node_id), None, ) surface_ids = ( [ str(surface_id).strip() for surface_id in node.config.get("view_surface_ids") or () if str(surface_id).strip() ] if node is not None else [] ) return WorkflowViewContextResponse( view_id=revision.view_id, revision_id=revision.view_revision_id, visible_surface_ids=list(dict.fromkeys(surface_ids)), step_id=step.id if step is not None else None, node_id=node.id if node is not None else None, ) def _node(graph: WorkflowGraph, node_id: str) -> WorkflowNode: node = next((item for item in graph.nodes if item.id == node_id), None) if node is None: raise WorkflowConflictError(f"Workflow node {node_id!r} no longer exists.") return node def _runtime_graph(revision: WorkflowDefinitionRevision) -> WorkflowGraph: try: return materialize_runtime_graph(WorkflowGraph.model_validate(revision.graph)) except BpmnGraphError as exc: raise WorkflowConflictError(str(exc)) from exc def _next_node_id( graph: WorkflowGraph, source_id: str, port: str, ) -> str | None: outgoing = [edge for edge in graph.edges if edge.source == source_id] exact = [edge for edge in outgoing if edge.source_port == port] if len(exact) == 1: return exact[0].target if not exact and len(outgoing) == 1: return outgoing[0].target if not exact: return None raise WorkflowConflictError( f"Workflow node {source_id!r} has multiple {port!r} transitions." ) def _action_port( step: WorkflowInstanceStep, action: str, ) -> str: if step.node_type == "workflow.review": return { "approve": "approved", "complete": "approved", "reject": "rejected", }.get(action, "changes") if step.node_type == "workflow.wait": return "resumed" if step.node_type == "workflow.dataflow": if action == "reject": return "failure" return str(step.handoff.get("suggested_port") or "success") if step.node_type == "workflow.capability": if action == "reject": return "failure" return str(step.handoff.get("suggested_port") or "success") return "output" def _dataflow_output( descriptor: DataflowRunDescriptor, ) -> dict[str, object]: return { "run_ref": descriptor.ref, "status": descriptor.status, "definition_hash": descriptor.definition_hash, "output_publication_ref": descriptor.output_publication_ref, "output_datasource_ref": descriptor.output_datasource_ref, "output_materialization_ref": descriptor.output_materialization_ref, "input_row_count": descriptor.input_row_count, "output_row_count": descriptor.output_row_count, "diagnostics": list(descriptor.metadata.get("diagnostics") or []), } def _dataflow_action_url(pipeline_ref: str, run_ref: str) -> str: pipeline_id = pipeline_ref.removeprefix("pipeline:") return f"/dataflow?pipelineId={pipeline_id}&runRef={run_ref}" def _require_runtime_dependencies( graph: WorkflowGraph, *, principal: ApiPrincipal, registry: object | None, ) -> None: if any(node.type == "workflow.dataflow" for node in graph.nodes): if dataflow_run_lifecycle(registry) is None: raise WorkflowConflictError( "This Workflow requires the optional Dataflow module." ) if not has_scope(principal, DATAFLOW_RUN_SCOPE): raise WorkflowConflictError( "Starting this Workflow requires dataflow:pipeline:run." ) for node in graph.nodes: if node.type == "workflow.capability": _capability_action_definition( node, principal=principal, registry=registry, ) def _authorization_payload( principal: ApiPrincipal, *, graph: WorkflowGraph, registry: object | None, ) -> dict[str, object]: principal_ref = principal.to_platform_principal() scopes = required_instance_scopes( graph, principal=principal, registry=registry, ) return { "contract_version": "1", "subject_kind": ( "service_account" if principal_ref.service_account_id else "delegated_user" ), "account_id": principal_ref.account_id, "membership_id": principal_ref.membership_id, "service_account_id": principal_ref.service_account_id, "grant_scopes": list(scopes), "authorization_ref": None, } def required_instance_scopes( graph: WorkflowGraph, *, principal: ApiPrincipal, registry: object | None, ) -> tuple[str, ...]: """Return scopes pinned into an instance or trigger authorization artifact.""" scopes = {INSTANCE_START_SCOPE} if any(node.type == "workflow.dataflow" for node in graph.nodes): scopes.add(DATAFLOW_RUN_SCOPE) for node in graph.nodes: if node.type != "workflow.capability": continue _provider, definition = _capability_action_definition( node, principal=principal, registry=registry, ) scopes.update(definition.required_scopes) return tuple(sorted(scopes)) def _resolve_instance_principal( session: Session, *, instance: WorkflowInstance, registry: object | None, ) -> ApiPrincipal | None: provider = automation_principal_provider(registry) if provider is None: return None value = dict(instance.authorization_) common = { "tenant_id": instance.tenant_id, "authorization_ref": str( value.get("authorization_ref") or f"workflow-instance:{instance.id}" ), "grant_scopes": tuple(str(scope) for scope in value.get("grant_scopes") or ()), "context": { "workflow_instance_ref": f"workflow-instance:{instance.id}", "definition_ref": (f"workflow-definition:{instance.definition_id}"), }, } try: if value.get("subject_kind") == "service_account": request = AutomationPrincipalRequest.service_account( service_account_id=str(value.get("service_account_id") or ""), **common, ) else: request = AutomationPrincipalRequest.delegated_user( account_id=str(value.get("account_id") or ""), membership_id=str(value.get("membership_id") or ""), **common, ) except ValueError as exc: instance.authorization_ = { **value, "last_resolution": { "allowed": False, "reason": str(exc), }, "resolved_at": utcnow().isoformat(), } return None resolution = provider.resolve_automation_principal( session, request=request, ) instance.authorization_ = { **value, "last_resolution": dict(resolution.provenance), "resolved_at": utcnow().isoformat(), } return ( resolution.principal if resolution.allowed and isinstance(resolution.principal, ApiPrincipal) else None ) def _notify_handoff( session: Session, *, registry: object | None, instance: WorkflowInstance, step: WorkflowInstanceStep, subject: str, ) -> None: provider = notification_dispatch_provider(registry) account_id = str(instance.authorization_.get("account_id") or "").strip() if provider is None or not account_id: return try: provider.enqueue_notification( session, NotificationDispatchRequest( tenant_id=instance.tenant_id, source_module="workflow", source_resource_type="workflow_instance", source_resource_id=instance.id, event_kind="workflow.handoff.required", recipient_type="account", recipient_id=account_id, subject=subject, action_url=( f"/workflow?definition={instance.definition_id}&run={instance.id}" ), payload={ "instance_id": instance.id, "step_id": step.id, "handoff": dict(step.handoff), }, ), ) except Exception: logger.warning( "Workflow handoff notification enqueue failed for instance %s", instance.id, exc_info=True, ) class SqlWorkflowRuntimeWorker: def __init__(self, *, registry: object | None = None) -> None: self._registry = registry self._standards_reconciled_tenants: set[str | None] = set() def reconcile_pending( self, session: object, *, tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> Mapping[str, object]: if not isinstance(session, Session): raise TypeError("Workflow reconciliation requires a Session.") standards: Mapping[str, object] | None = None if tenant_id not in self._standards_reconciled_tenants: from govoplan_workflow_engine.backend.contributions import ( reconcile_workflow_definition_contributions, ) standards = reconcile_workflow_definition_contributions( session, registry=self._registry, tenant_ids=((tenant_id,) if tenant_id else ()), ) if int(standards.get("blocked") or 0) == 0: self._standards_reconciled_tenants.add(tenant_id) runtime = reconcile_pending_instances( session, registry=self._registry, tenant_id=tenant_id, limit=limit, ) from govoplan_workflow_engine.backend.triggers import dispatch_due_work triggers = dispatch_due_work( session, registry=self._registry, tenant_id=tenant_id, now=now, limit=limit, ) return { **runtime, "triggers": triggers, **({"standards": standards} if standards is not None else {}), } __all__ = [ "SqlWorkflowRuntimeWorker", "cancel_instance", "get_instance", "instance_response", "list_instances", "reconcile_instance", "reconcile_pending_instances", "required_instance_scopes", "resolve_step", "start_instance", ]