diff --git a/README.md b/README.md index 782e75c..fee1131 100644 --- a/README.md +++ b/README.md @@ -20,10 +20,13 @@ fields, connected nodes, and permitted loops for correction and retry paths. Dataflow uses the same graph contract with its own acyclic transformation library. -The current slice deliberately stops before executing process instances. -Instance state, resumable transitions, human activities, retries, and event -subscriptions must build on the pinned definition and versioned module -capabilities rather than bypassing them. +The executable runtime persists revision-pinned instances, append-only +transition evidence, resumable human handoffs, retries, cancellation, and +stable external output references. Dataflow nodes enqueue work through +Dataflow's lifecycle capability; Workflow never imports Dataflow internals. +Core's periodic worker reconciles linked runs after re-resolving the stored +automation principal, while the operator surface exposes progress, review +actions, evidence references, and direct navigation to Dataflow results. Definitions can be complete flows or non-runnable templates at system, tenant, group, or user scope. Policy resolves whether a definition can be @@ -33,9 +36,10 @@ and records its hash, node-library version, source scope, actor, Policy decision, and effective ancestor limits. The start-node library distinguishes explicit user, API, scheduled, event, and -parent-workflow starts. These are definition contracts only until the -resumable Workflow instance runtime is implemented; the UI reports that -limitation rather than presenting automation as operational. +parent-workflow starts. Manual starts and Dataflow/human handoffs are +operational. The other trigger and generic capability nodes remain explicit +definition contracts until their event/schedule dispatchers and versioned +operation providers are implemented. See [docs/CONCEPT.md](docs/CONCEPT.md) for the complete module concept. diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md index 061cc9e..16d0f2d 100644 --- a/docs/CONCEPT.md +++ b/docs/CONCEPT.md @@ -104,19 +104,23 @@ The first executable slice now provides: - trigger, activity, review, decision, wait, module-action, Dataflow, and outcome nodes - API discovery and validation endpoints +- revision-pinned, idempotent Workflow instances +- persisted steps and append-only transition evidence +- durable Dataflow handoff, progress reconciliation, output references, + retries, cancellation, and warning/review paths +- manual activity, review, and wait handoffs with comments and evidence +- a worker capability with current-authorization rechecks +- an operator dialog for starting, inspecting, and advancing instances The next execution slices should provide: - static workflow definition registration from configuration packages -- create/read/list workflow instances -- transition execution with permission checks +- event, API, schedule, and parent-workflow start dispatchers - guard hooks implemented through capability calls -- command execution records with retry/manual-resolution state +- registry-driven generic module-action execution records - action/effect previews for transitions that call other modules -- idempotency keys for command execution - explicit blocked, retryable, quarantined, manual-required, and compensation-required states -- basic WebUI instance detail and definition viewer - dashboard summary provider - event emission and audit integration @@ -149,12 +153,16 @@ details. ## Data Model Sketch -Candidate tables: +Current tables: - `workflow_definitions` -- `workflow_definition_versions` +- `workflow_definition_revisions` - `workflow_instances` -- `workflow_transition_history` +- `workflow_instance_steps` +- `workflow_instance_events` + +Future generic action execution and timers may add: + - `workflow_command_records` - `workflow_timers` @@ -163,16 +171,14 @@ reference the exact version used at start. ## WebUI -Initial route contributions: +Current route contribution: - `/workflow` -- `/workflow/instances/:instanceId` -- `/workflow/definitions/:definitionId` -The UI should show current state, available transitions, pending commands, -failed handoffs, audit trace, and linked subject records. It should not import -case/task/template components directly; panels are contributed through core UI -extension points. +The route combines the definition editor and a fixed run dialog showing current +state, available transitions, failed handoffs, comments/evidence, immutable +event history, and linked Dataflow results. It does not import Dataflow or other +domain UI components. ## Tests diff --git a/src/govoplan_workflow/backend/db/__init__.py b/src/govoplan_workflow/backend/db/__init__.py index 58df07b..4d74ee9 100644 --- a/src/govoplan_workflow/backend/db/__init__.py +++ b/src/govoplan_workflow/backend/db/__init__.py @@ -1,9 +1,15 @@ from govoplan_workflow.backend.db.models import ( WorkflowDefinition, WorkflowDefinitionRevision, + WorkflowInstance, + WorkflowInstanceEvent, + WorkflowInstanceStep, ) __all__ = [ "WorkflowDefinition", "WorkflowDefinitionRevision", + "WorkflowInstance", + "WorkflowInstanceEvent", + "WorkflowInstanceStep", ] diff --git a/src/govoplan_workflow/backend/db/models.py b/src/govoplan_workflow/backend/db/models.py index 7a0413e..fca752c 100644 --- a/src/govoplan_workflow/backend/db/models.py +++ b/src/govoplan_workflow/backend/db/models.py @@ -140,6 +140,11 @@ class WorkflowDefinition(Base, TimestampMixin): cascade="all, delete-orphan", order_by="WorkflowDefinitionRevision.revision", ) + instances: Mapped[list["WorkflowInstance"]] = relationship( + back_populates="definition", + cascade="all, delete-orphan", + order_by="WorkflowInstance.created_at", + ) class WorkflowDefinitionRevision(Base, TimestampMixin): @@ -188,8 +193,244 @@ class WorkflowDefinitionRevision(Base, TimestampMixin): definition: Mapped[WorkflowDefinition] = relationship(back_populates="revisions") +class WorkflowInstance(Base, TimestampMixin): + __tablename__ = "workflow_instances" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "definition_id", + "idempotency_key", + name="uq_workflow_instance_idempotency", + ), + Index( + "ix_workflow_instances_tenant_status", + "tenant_id", + "status", + ), + Index( + "ix_workflow_instances_reconcile", + "status", + "updated_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + definition_id: Mapped[str] = mapped_column( + ForeignKey("workflow_definitions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + definition_revision_id: Mapped[str] = mapped_column( + ForeignKey("workflow_definition_revisions.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ) + status: Mapped[str] = mapped_column( + String(30), + default="running", + nullable=False, + index=True, + ) + idempotency_key: Mapped[str] = mapped_column( + String(255), + nullable=False, + index=True, + ) + correlation_id: Mapped[str | None] = mapped_column( + String(128), + nullable=True, + index=True, + ) + current_step_id: Mapped[str | None] = mapped_column( + String(36), + nullable=True, + index=True, + ) + input_: Mapped[dict[str, Any]] = mapped_column( + "input", + JSON, + default=dict, + nullable=False, + ) + context_: Mapped[dict[str, Any]] = mapped_column( + "context", + JSON, + default=dict, + nullable=False, + ) + output_: Mapped[dict[str, Any]] = mapped_column( + "output", + JSON, + default=dict, + nullable=False, + ) + authorization_: Mapped[dict[str, Any]] = mapped_column( + "authorization", + JSON, + default=dict, + nullable=False, + ) + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + cancellation_requested_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + + definition: Mapped[WorkflowDefinition] = relationship( + back_populates="instances" + ) + steps: Mapped[list["WorkflowInstanceStep"]] = relationship( + back_populates="instance", + cascade="all, delete-orphan", + order_by="WorkflowInstanceStep.sequence", + ) + events: Mapped[list["WorkflowInstanceEvent"]] = relationship( + back_populates="instance", + cascade="all, delete-orphan", + order_by="WorkflowInstanceEvent.sequence", + ) + + +class WorkflowInstanceStep(Base, TimestampMixin): + __tablename__ = "workflow_instance_steps" + __table_args__ = ( + UniqueConstraint( + "instance_id", + "sequence", + name="uq_workflow_instance_step_sequence", + ), + Index( + "ix_workflow_instance_steps_tenant_status", + "tenant_id", + "status", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + instance_id: Mapped[str] = mapped_column( + ForeignKey("workflow_instances.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + sequence: Mapped[int] = mapped_column(Integer, nullable=False) + node_id: Mapped[str] = mapped_column(String(120), nullable=False, index=True) + node_type: Mapped[str] = mapped_column(String(120), nullable=False, index=True) + status: Mapped[str] = mapped_column( + String(30), + nullable=False, + index=True, + ) + attempt: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + idempotency_key: Mapped[str] = mapped_column( + String(255), + nullable=False, + ) + input_: Mapped[dict[str, Any]] = mapped_column( + "input", + JSON, + default=dict, + nullable=False, + ) + output_: Mapped[dict[str, Any]] = mapped_column( + "output", + JSON, + default=dict, + nullable=False, + ) + handoff: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + external_ref: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + index=True, + ) + started_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + finished_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + completed_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + ) + + instance: Mapped[WorkflowInstance] = relationship(back_populates="steps") + + +class WorkflowInstanceEvent(Base): + __tablename__ = "workflow_instance_events" + __table_args__ = ( + UniqueConstraint( + "instance_id", + "sequence", + name="uq_workflow_instance_event_sequence", + ), + Index( + "ix_workflow_instance_events_tenant_created", + "tenant_id", + "created_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + instance_id: Mapped[str] = mapped_column( + ForeignKey("workflow_instances.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + step_id: Mapped[str | None] = mapped_column( + String(36), + nullable=True, + index=True, + ) + sequence: Mapped[int] = mapped_column(Integer, nullable=False) + kind: Mapped[str] = mapped_column(String(120), nullable=False, index=True) + actor_id: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + payload: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + ) + + instance: Mapped[WorkflowInstance] = relationship(back_populates="events") + + __all__ = [ "WorkflowDefinition", "WorkflowDefinitionRevision", + "WorkflowInstance", + "WorkflowInstanceEvent", + "WorkflowInstanceStep", "new_uuid", ] diff --git a/src/govoplan_workflow/backend/governance.py b/src/govoplan_workflow/backend/governance.py index 92621b0..ab03d02 100644 --- a/src/govoplan_workflow/backend/governance.py +++ b/src/govoplan_workflow/backend/governance.py @@ -12,6 +12,7 @@ from govoplan_core.core.policy import ( PolicySourceStep, definition_governance_policy, ) +from govoplan_core.core.workflows import workflow_runtime_worker from govoplan_workflow.backend.db.models import WorkflowDefinition @@ -127,6 +128,7 @@ def definition_governance_payload( principal: ApiPrincipal, registry: object | None, ) -> dict[str, object]: + runtime_available = workflow_runtime_worker(registry) is not None actions = { action: definition_decision( definition, @@ -151,10 +153,11 @@ def definition_governance_payload( "derived_from_hash": definition.derived_from_hash, "derivation_provenance": dict(definition.derivation_provenance), "actions": actions, - "automation_runtime_available": False, + "automation_runtime_available": runtime_available, "automation_runtime_reason": ( - "Workflow start definitions are persisted and governed, but " - "automatic instance dispatch requires the Workflow runtime." + None + if runtime_available + else "Automatic reconciliation requires the Workflow runtime worker." ), } diff --git a/src/govoplan_workflow/backend/instance_service.py b/src/govoplan_workflow/backend/instance_service.py new file mode 100644 index 0000000..7602825 --- /dev/null +++ b/src/govoplan_workflow/backend/instance_service.py @@ -0,0 +1,1484 @@ +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime +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 ( + AutomationInvocation, + AutomationPrincipalRequest, + 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.db.base import utcnow +from govoplan_workflow.backend.db.models import ( + WorkflowDefinition, + WorkflowDefinitionRevision, + WorkflowInstance, + WorkflowInstanceEvent, + WorkflowInstanceStep, +) +from govoplan_workflow.backend.governance import require_definition_action +from govoplan_workflow.backend.schemas import ( + WorkflowGraph, + WorkflowInstanceEventResponse, + WorkflowInstanceResponse, + WorkflowInstanceStartRequest, + WorkflowInstanceStepResponse, + WorkflowNode, + WorkflowStepActionRequest, +) +from govoplan_workflow.backend.service import ( + WorkflowConflictError, + WorkflowNotFoundError, + get_definition, + get_definition_revision, +) + + +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, +) -> 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, + ) + graph = WorkflowGraph.model_validate(revision.graph) + _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 + ): + 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="manual") + now = utcnow() + instance = WorkflowInstance( + tenant_id=tenant_id, + definition_id=definition.id, + definition_revision_id=revision.id, + status="running", + 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, + ), + 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, + "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 or step.node_type != "workflow.dataflow": + 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 = WorkflowGraph.model_validate(revision.graph) + node = _node(graph, step.node_id) + if not step.external_ref: + _start_dataflow_step( + session, + instance=instance, + step=step, + node=node, + principal=principal, + registry=registry, + ) + 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: + _set_failure_handoff( + session, + instance=instance, + step=step, + message=str(exc), + ) + return True + if descriptor is None: + _set_failure_handoff( + session, + instance=instance, + step=step, + message="The linked Dataflow run no longer exists.", + ) + return True + 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": + _set_failure_handoff( + session, + instance=instance, + step=step, + message="The linked Dataflow run was cancelled.", + state="cancelled", + ) + return True + _set_failure_handoff( + session, + instance=instance, + step=step, + 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 = WorkflowGraph.model_validate(revision.graph) + 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 == "changes": + step.handoff = { + **dict(step.handoff), + "state": "changes_requested", + "last_comment": payload.comment, + } + return instance + if payload.action == "retry": + if step.node_type != "workflow.dataflow": + raise WorkflowConflictError( + "Only failed 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), + } + 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 and 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, + ) + 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, + limit: int = 50, +) -> dict[str, object]: + instances = list( + session.scalars( + select(WorkflowInstance) + .where( + WorkflowInstance.status.in_(("running", "waiting")), + WorkflowInstance.current_step_id.is_not(None), + ) + .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: + step = _current_step(session, instance) + if step is None or step.node_type != "workflow.dataflow": + summary["waiting"] = int(summary["waiting"]) + 1 + continue + principal = _resolve_instance_principal( + session, + instance=instance, + registry=registry, + ) + if principal is None: + summary["skipped"] = int(summary["skipped"]) + 1 + 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 + 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." + ) + return WorkflowInstanceResponse( + id=instance.id, + definition_id=instance.definition_id, + definition_name=definition.name, + definition_revision=revision.revision, + definition_hash=revision.content_hash, + 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, + ) + return + if node.type in { + "workflow.activity", + "workflow.review", + "workflow.wait", + }: + _set_human_handoff( + session, + instance=instance, + step=step, + node=node, + registry=registry, + ) + 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 _start_dataflow_step( + session: Session, + *, + instance: WorkflowInstance, + step: WorkflowInstanceStep, + node: WorkflowNode, + principal: ApiPrincipal, + registry: object | 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): + _set_failure_handoff( + session, + instance=instance, + step=step, + message="Dataflow revision and row limit must be integers.", + ) + return + if not pipeline_ref or revision < 1: + _set_failure_handoff( + session, + instance=instance, + step=step, + 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: + _set_failure_handoff( + session, + instance=instance, + step=step, + 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_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 _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() + + +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: + raise WorkflowConflictError( + f"Workflow has no {kind} start node." + ) + return node + + +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 _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") + 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 not any(node.type == "workflow.dataflow" for node in graph.nodes): + return + 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." + ) + + +def _authorization_payload( + principal: ApiPrincipal, + *, + graph: WorkflowGraph, +) -> dict[str, object]: + principal_ref = principal.to_platform_principal() + scopes = {INSTANCE_START_SCOPE} + if any(node.type == "workflow.dataflow" for node in graph.nodes): + scopes.add(DATAFLOW_RUN_SCOPE) + return { + "contract_version": "1", + "subject_kind": ( + "service_account" + if principal_ref.service_account_id + else "delegated_user" + ), + "account_id": principal_ref.account_id, + "membership_id": principal_ref.membership_id, + "service_account_id": principal_ref.service_account_id, + "grant_scopes": sorted(scopes), + "authorization_ref": None, + } + + +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="/workflow", + 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 + + def reconcile_pending( + self, + session: object, + *, + now: datetime | None = None, + limit: int = 50, + ) -> Mapping[str, object]: + del now + if not isinstance(session, Session): + raise TypeError("Workflow reconciliation requires a Session.") + return reconcile_pending_instances( + session, + registry=self._registry, + limit=limit, + ) + + +__all__ = [ + "SqlWorkflowRuntimeWorker", + "cancel_instance", + "get_instance", + "instance_response", + "list_instances", + "reconcile_instance", + "reconcile_pending_instances", + "resolve_step", + "start_instance", +] diff --git a/src/govoplan_workflow/backend/manifest.py b/src/govoplan_workflow/backend/manifest.py index 8402aa7..badf67a 100644 --- a/src/govoplan_workflow/backend/manifest.py +++ b/src/govoplan_workflow/backend/manifest.py @@ -4,6 +4,7 @@ from pathlib import Path from govoplan_core.core.access import ( CAPABILITY_ACCESS_DIRECTORY, + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) @@ -28,8 +29,14 @@ from govoplan_core.core.modules import ( from govoplan_core.core.policy import ( CAPABILITY_POLICY_DEFINITION_GOVERNANCE, ) +from govoplan_core.core.notifications import ( + CAPABILITY_NOTIFICATIONS_DISPATCH, +) 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_RUNTIME_WORKER, +) from govoplan_core.db.base import Base from govoplan_workflow.backend.db import models as workflow_models @@ -123,6 +130,14 @@ def _router(context: ModuleContext): return router +def _runtime_worker(context: ModuleContext): + from govoplan_workflow.backend.instance_service import ( + SqlWorkflowRuntimeWorker, + ) + + return SqlWorkflowRuntimeWorker(registry=context.registry) + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -141,9 +156,11 @@ manifest = ModuleManifest( optional_capabilities=( CAPABILITY_ACCESS_DIRECTORY, CAPABILITY_ACCESS_REFERENCE_OPTIONS, + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_DATAFLOW_RUN_LIFECYCLE, + CAPABILITY_NOTIFICATIONS_DISPATCH, CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_VIEWS_RESOLVER, ), @@ -151,6 +168,7 @@ manifest = ModuleManifest( ModuleInterfaceProvider(name="workflow.definition_graph", version="0.1.0"), ModuleInterfaceProvider(name="workflow.node_library", version="0.1.0"), ModuleInterfaceProvider(name="workflow.definition_catalogue", version="0.1.0"), + ModuleInterfaceProvider(name="workflow.runtime_worker", version=MODULE_VERSION), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -165,6 +183,18 @@ manifest = ModuleManifest( version_max_exclusive="1.0.0", optional=True, ), + ModuleInterfaceRequirement( + name="auth.automation_principal", + version_min="0.1.0", + version_max_exclusive="1.0.0", + optional=True, + ), + ModuleInterfaceRequirement( + name=CAPABILITY_NOTIFICATIONS_DISPATCH, + version_min="0.1.0", + version_max_exclusive="1.0.0", + optional=True, + ), ModuleInterfaceRequirement( name="policy.definition_governance", version_min="0.1.0", @@ -211,12 +241,18 @@ manifest = ModuleManifest( ), ), route_factory=_router, + capability_factories={ + CAPABILITY_WORKFLOW_RUNTIME_WORKER: _runtime_worker, + }, migration_spec=MigrationSpec( module_id=MODULE_ID, metadata=Base.metadata, script_location=str(Path(__file__).with_name("migrations") / "versions"), retirement_supported=True, retirement_provider=drop_table_retirement_provider( + workflow_models.WorkflowInstanceEvent, + workflow_models.WorkflowInstanceStep, + workflow_models.WorkflowInstance, workflow_models.WorkflowDefinitionRevision, workflow_models.WorkflowDefinition, label="Workflow", @@ -230,6 +266,9 @@ manifest = ModuleManifest( persistent_table_uninstall_guard( workflow_models.WorkflowDefinition, workflow_models.WorkflowDefinitionRevision, + workflow_models.WorkflowInstance, + workflow_models.WorkflowInstanceStep, + workflow_models.WorkflowInstanceEvent, label="Workflow", ), ), diff --git a/src/govoplan_workflow/backend/migrations/versions/d8f2a5c7e1b4_v0114_workflow_runtime.py b/src/govoplan_workflow/backend/migrations/versions/d8f2a5c7e1b4_v0114_workflow_runtime.py new file mode 100644 index 0000000..a7479a9 --- /dev/null +++ b/src/govoplan_workflow/backend/migrations/versions/d8f2a5c7e1b4_v0114_workflow_runtime.py @@ -0,0 +1,213 @@ +"""v0.1.14 Workflow instances and resumable handoffs + +Revision ID: d8f2a5c7e1b4 +Revises: c6d8f1a3e5b7 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "d8f2a5c7e1b4" +down_revision = "c6d8f1a3e5b7" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "workflow_instances", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("definition_id", sa.String(length=36), nullable=False), + sa.Column( + "definition_revision_id", + sa.String(length=36), + nullable=False, + ), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("idempotency_key", sa.String(length=255), nullable=False), + sa.Column("correlation_id", sa.String(length=128), nullable=True), + sa.Column("current_step_id", sa.String(length=36), nullable=True), + sa.Column("input", sa.JSON(), nullable=False), + sa.Column("context", sa.JSON(), nullable=False), + sa.Column("output", sa.JSON(), nullable=False), + sa.Column("authorization", sa.JSON(), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "cancellation_requested_at", + sa.DateTime(timezone=True), + nullable=True, + ), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["definition_id"], + ["workflow_definitions.id"], + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["definition_revision_id"], + ["workflow_definition_revisions.id"], + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "tenant_id", + "definition_id", + "idempotency_key", + name="uq_workflow_instance_idempotency", + ), + ) + for column in ( + "tenant_id", + "definition_id", + "definition_revision_id", + "status", + "idempotency_key", + "correlation_id", + "current_step_id", + "created_by", + ): + op.create_index( + op.f(f"ix_workflow_instances_{column}"), + "workflow_instances", + [column], + unique=False, + ) + op.create_index( + "ix_workflow_instances_tenant_status", + "workflow_instances", + ["tenant_id", "status"], + unique=False, + ) + op.create_index( + "ix_workflow_instances_reconcile", + "workflow_instances", + ["status", "updated_at"], + unique=False, + ) + + op.create_table( + "workflow_instance_steps", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("instance_id", sa.String(length=36), nullable=False), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("node_id", sa.String(length=120), nullable=False), + sa.Column("node_type", sa.String(length=120), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("attempt", sa.Integer(), nullable=False), + sa.Column("idempotency_key", sa.String(length=255), nullable=False), + sa.Column("input", sa.JSON(), nullable=False), + sa.Column("output", sa.JSON(), nullable=False), + sa.Column("handoff", sa.JSON(), nullable=False), + sa.Column("external_ref", sa.String(length=500), nullable=True), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("completed_by", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["instance_id"], + ["workflow_instances.id"], + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "instance_id", + "sequence", + name="uq_workflow_instance_step_sequence", + ), + ) + for column in ( + "tenant_id", + "instance_id", + "node_id", + "node_type", + "status", + "external_ref", + ): + op.create_index( + op.f(f"ix_workflow_instance_steps_{column}"), + "workflow_instance_steps", + [column], + unique=False, + ) + op.create_index( + "ix_workflow_instance_steps_tenant_status", + "workflow_instance_steps", + ["tenant_id", "status"], + unique=False, + ) + + op.create_table( + "workflow_instance_events", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("instance_id", sa.String(length=36), nullable=False), + sa.Column("step_id", sa.String(length=36), nullable=True), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("kind", sa.String(length=120), nullable=False), + sa.Column("actor_id", sa.String(length=255), nullable=True), + sa.Column("payload", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["instance_id"], + ["workflow_instances.id"], + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "instance_id", + "sequence", + name="uq_workflow_instance_event_sequence", + ), + ) + for column in ( + "tenant_id", + "instance_id", + "step_id", + "kind", + "actor_id", + ): + op.create_index( + op.f(f"ix_workflow_instance_events_{column}"), + "workflow_instance_events", + [column], + unique=False, + ) + op.create_index( + "ix_workflow_instance_events_tenant_created", + "workflow_instance_events", + ["tenant_id", "created_at"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + "ix_workflow_instance_events_tenant_created", + table_name="workflow_instance_events", + ) + op.drop_table("workflow_instance_events") + op.drop_index( + "ix_workflow_instance_steps_tenant_status", + table_name="workflow_instance_steps", + ) + op.drop_table("workflow_instance_steps") + op.drop_index( + "ix_workflow_instances_reconcile", + table_name="workflow_instances", + ) + op.drop_index( + "ix_workflow_instances_tenant_status", + table_name="workflow_instances", + ) + op.drop_table("workflow_instances") diff --git a/src/govoplan_workflow/backend/node_library.py b/src/govoplan_workflow/backend/node_library.py index 666ca8e..05fc761 100644 --- a/src/govoplan_workflow/backend/node_library.py +++ b/src/govoplan_workflow/backend/node_library.py @@ -243,6 +243,12 @@ WORKFLOW_NODE_TYPES = ( input_ports=(DefinitionPort(id="input", label="Input"),), output_ports=( DefinitionPort(id="success", label="Success", required=False), + DefinitionPort(id="warning", label="Warning", required=False), + DefinitionPort( + id="review_required", + label="Review required", + required=False, + ), DefinitionPort(id="failure", label="Failure", required=False), ), config_fields=( @@ -294,6 +300,12 @@ WORKFLOW_NODE_TYPES = ( input_ports=(DefinitionPort(id="input", label="Input"),), output_ports=( DefinitionPort(id="success", label="Success", required=False), + DefinitionPort(id="warning", label="Warning", required=False), + DefinitionPort( + id="review_required", + label="Review required", + required=False, + ), DefinitionPort(id="failure", label="Failure", required=False), ), config_fields=( @@ -303,9 +315,59 @@ WORKFLOW_NODE_TYPES = ( kind="dataflow", required=True, ), + DefinitionConfigField( + id="revision", + label="Pinned revision", + kind="number", + required=True, + ), + DefinitionConfigField( + id="environment", + label="Environment", + kind="select", + required=True, + options=( + ("development", "Development"), + ("staging", "Staging"), + ("production", "Production"), + ), + ), + DefinitionConfigField( + id="row_limit", + label="Output row limit", + kind="number", + required=True, + ), + DefinitionConfigField( + id="publication_target_ref", + label="Publication datasource", + kind="text", + description=( + "Optional stable Datasource target for materialized " + "output." + ), + ), + DefinitionConfigField( + id="warning_policy", + label="Warnings", + kind="select", + required=True, + options=( + ("review", "Require review"), + ("continue", "Continue"), + ), + ), DefinitionConfigField(id="input_mapping", label="Input mapping", kind="mapping"), ), - default_config={"pipeline_ref": "", "input_mapping": {}}, + default_config={ + "pipeline_ref": "", + "revision": 1, + "environment": "development", + "row_limit": 500, + "publication_target_ref": "", + "warning_policy": "review", + "input_mapping": {}, + }, ), DefinitionNodeType( type="workflow.end.completed", diff --git a/src/govoplan_workflow/backend/router.py b/src/govoplan_workflow/backend/router.py index 20674b7..f5a4ee8 100644 --- a/src/govoplan_workflow/backend/router.py +++ b/src/govoplan_workflow/backend/router.py @@ -24,6 +24,9 @@ from govoplan_workflow.backend.manifest import ( ADMIN_SCOPE, DEFINITION_READ_SCOPE, DEFINITION_WRITE_SCOPE, + INSTANCE_READ_SCOPE, + INSTANCE_START_SCOPE, + INSTANCE_TRANSITION_SCOPE, ) from govoplan_workflow.backend.node_library import WORKFLOW_GRAPH_LIBRARY from govoplan_workflow.backend.schemas import ( @@ -40,9 +43,22 @@ from govoplan_workflow.backend.schemas import ( WorkflowDiagnosticResponse, WorkflowGraphValidationRequest, WorkflowGraphValidationResponse, + WorkflowInstanceListResponse, + WorkflowInstanceResponse, + WorkflowInstanceStartRequest, WorkflowNodeLibraryResponse, WorkflowNodeTypeResponse, WorkflowPortResponse, + WorkflowStepActionRequest, +) +from govoplan_workflow.backend.instance_service import ( + cancel_instance, + get_instance, + instance_response, + list_instances, + reconcile_instance, + resolve_step, + start_instance, ) from govoplan_workflow.backend.runtime import get_registry from govoplan_workflow.backend.service import ( @@ -159,6 +175,35 @@ def _audit( ) +def _audit_instance( + session: Session, + principal: ApiPrincipal, + *, + action: str, + instance_id: str, + details: dict[str, object], +) -> None: + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action=action, + object_type="workflow_instance", + object_id=instance_id, + details=details, + ) + + +def _require_instance_view(instance, principal: ApiPrincipal) -> None: + require_definition_action( + instance.definition, + principal=principal, + registry=get_registry(), + action="view", + ) + + @router.get("/node-types", response_model=WorkflowNodeLibraryResponse) def api_node_types( principal: ApiPrincipal = Depends(get_api_principal), @@ -324,6 +369,242 @@ def api_list_definitions( ) +@router.get("/instances", response_model=WorkflowInstanceListResponse) +def api_list_instances( + definition_id: str | None = None, + limit: int = Query(default=100, ge=1, le=200), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceListResponse: + _require_any_scope(principal, INSTANCE_READ_SCOPE, ADMIN_SCOPE) + try: + instances = [ + instance + for instance in list_instances( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + limit=limit, + ) + if definition_decision( + instance.definition, + principal=principal, + registry=get_registry(), + action="view", + ).allowed + ] + return WorkflowInstanceListResponse( + instances=[ + instance_response(session, instance) + for instance in instances + ] + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + + +@router.post( + "/definitions/{definition_id}/instances", + response_model=WorkflowInstanceResponse, + status_code=status.HTTP_201_CREATED, +) +def api_start_instance( + definition_id: str, + payload: WorkflowInstanceStartRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceResponse: + _require_any_scope(principal, INSTANCE_START_SCOPE, ADMIN_SCOPE) + try: + instance, replayed = start_instance( + session, + tenant_id=principal.tenant_id, + definition_id=definition_id, + actor_id=_actor_id(principal), + principal=principal, + registry=get_registry(), + payload=payload, + ) + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit_instance( + session, + principal, + action=( + "workflow.instance.replayed" + if replayed + else "workflow.instance.started" + ), + instance_id=instance.id, + details={ + "definition_id": instance.definition_id, + "definition_revision_id": instance.definition_revision_id, + "idempotency_key": instance.idempotency_key, + }, + ) + response = instance_response(session, instance, replayed=replayed) + session.commit() + return response + + +@router.get( + "/instances/{instance_id}", + response_model=WorkflowInstanceResponse, +) +def api_get_instance( + instance_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceResponse: + _require_any_scope(principal, INSTANCE_READ_SCOPE, ADMIN_SCOPE) + try: + instance = get_instance( + session, + tenant_id=principal.tenant_id, + instance_id=instance_id, + ) + _require_instance_view(instance, principal) + return instance_response(session, instance) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + + +@router.post( + "/instances/{instance_id}/reconcile", + response_model=WorkflowInstanceResponse, +) +def api_reconcile_instance( + instance_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceResponse: + _require_any_scope(principal, INSTANCE_TRANSITION_SCOPE, ADMIN_SCOPE) + try: + instance = get_instance( + session, + tenant_id=principal.tenant_id, + instance_id=instance_id, + for_update=True, + ) + _require_instance_view(instance, principal) + changed = reconcile_instance( + session, + instance=instance, + principal=principal, + registry=get_registry(), + actor_id=_actor_id(principal), + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + if changed: + _audit_instance( + session, + principal, + action="workflow.instance.reconciled", + instance_id=instance.id, + details={ + "status": instance.status, + "current_step_id": instance.current_step_id, + }, + ) + response = instance_response(session, instance) + session.commit() + return response + + +@router.post( + "/instances/{instance_id}/steps/{step_id}/actions", + response_model=WorkflowInstanceResponse, +) +def api_resolve_instance_step( + instance_id: str, + step_id: str, + payload: WorkflowStepActionRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceResponse: + _require_any_scope(principal, INSTANCE_TRANSITION_SCOPE, ADMIN_SCOPE) + try: + existing = get_instance( + session, + tenant_id=principal.tenant_id, + instance_id=instance_id, + ) + _require_instance_view(existing, principal) + instance = resolve_step( + session, + tenant_id=principal.tenant_id, + instance_id=instance_id, + step_id=step_id, + actor_id=_actor_id(principal), + principal=principal, + registry=get_registry(), + payload=payload, + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit_instance( + session, + principal, + action=f"workflow.instance.{payload.action}", + instance_id=instance.id, + details={ + "step_id": step_id, + "status": instance.status, + }, + ) + response = instance_response(session, instance) + session.commit() + return response + + +@router.post( + "/instances/{instance_id}/cancel", + response_model=WorkflowInstanceResponse, +) +def api_cancel_instance( + instance_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceResponse: + _require_any_scope(principal, INSTANCE_TRANSITION_SCOPE, ADMIN_SCOPE) + try: + instance = get_instance( + session, + tenant_id=principal.tenant_id, + instance_id=instance_id, + ) + _require_instance_view(instance, principal) + instance = cancel_instance( + session, + tenant_id=principal.tenant_id, + instance_id=instance_id, + actor_id=_actor_id(principal), + principal=principal, + registry=get_registry(), + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + _audit_instance( + session, + principal, + action="workflow.instance.cancelled", + instance_id=instance.id, + details={"status": instance.status}, + ) + response = instance_response(session, instance) + session.commit() + return response + + @router.post( "/definitions", response_model=WorkflowDefinitionResponse, diff --git a/src/govoplan_workflow/backend/schemas.py b/src/govoplan_workflow/backend/schemas.py index 11102ab..6b5b20d 100644 --- a/src/govoplan_workflow/backend/schemas.py +++ b/src/govoplan_workflow/backend/schemas.py @@ -225,3 +225,99 @@ class WorkflowDefinitionActivateRequest(BaseModel): class WorkflowDefinitionDeleteResponse(BaseModel): deleted: bool definition_id: str + + +WorkflowInstanceStatus = Literal[ + "running", + "waiting", + "completed", + "failed", + "cancelled", +] +WorkflowStepStatus = Literal[ + "running", + "waiting", + "completed", + "failed", + "cancelled", + "superseded", +] + + +class WorkflowInstanceStartRequest(BaseModel): + idempotency_key: str = Field(min_length=1, max_length=255) + input: dict[str, Any] = Field(default_factory=dict) + correlation_id: str | None = Field(default=None, max_length=128) + + +class WorkflowStepActionRequest(BaseModel): + action: Literal[ + "complete", + "approve", + "changes", + "reject", + "resume", + "retry", + "cancel", + ] + output: dict[str, Any] = Field(default_factory=dict) + evidence: list[str] = Field(default_factory=list, max_length=100) + comment: str | None = Field(default=None, max_length=4_000) + + +class WorkflowInstanceStepResponse(BaseModel): + id: str + sequence: int + node_id: str + node_type: str + status: WorkflowStepStatus + attempt: int + input: dict[str, Any] + output: dict[str, Any] + handoff: dict[str, Any] + external_ref: str | None + started_at: datetime | None + finished_at: datetime | None + error: str | None + completed_by: str | None + created_at: datetime + updated_at: datetime + + +class WorkflowInstanceEventResponse(BaseModel): + id: str + sequence: int + step_id: str | None + kind: str + actor_id: str | None + payload: dict[str, Any] + created_at: datetime + + +class WorkflowInstanceResponse(BaseModel): + id: str + definition_id: str + definition_name: str + definition_revision: int + definition_hash: str + status: WorkflowInstanceStatus + idempotency_key: str + correlation_id: str | None + current_step_id: str | None + input: dict[str, Any] + context: dict[str, Any] + output: dict[str, Any] + started_at: datetime + finished_at: datetime | None + cancellation_requested_at: datetime | None + error: str | None + created_by: str | None + created_at: datetime + updated_at: datetime + steps: list[WorkflowInstanceStepResponse] + events: list[WorkflowInstanceEventResponse] + replayed: bool = False + + +class WorkflowInstanceListResponse(BaseModel): + instances: list[WorkflowInstanceResponse] diff --git a/tests/test_instance_service.py b/tests/test_instance_service.py new file mode 100644 index 0000000..25204ef --- /dev/null +++ b/tests/test_instance_service.py @@ -0,0 +1,482 @@ +from __future__ import annotations + +from dataclasses import replace +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import ( + CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, + PrincipalRef, +) +from govoplan_core.core.automation import AutomationPrincipalResolution +from govoplan_core.core.dataflows import ( + CAPABILITY_DATAFLOW_RUN_LIFECYCLE, + DataflowRunDescriptor, +) +from govoplan_core.db.base import Base +from govoplan_workflow.backend.db.models import ( + WorkflowDefinition, + WorkflowDefinitionRevision, + WorkflowInstance, + WorkflowInstanceEvent, + WorkflowInstanceStep, +) +from govoplan_workflow.backend.instance_service import ( + SqlWorkflowRuntimeWorker, + cancel_instance, + instance_response, + reconcile_instance, + resolve_step, + start_instance, +) +from govoplan_workflow.backend.schemas import ( + WorkflowDefinitionCreateRequest, + WorkflowEdge, + WorkflowGraph, + WorkflowInstanceStartRequest, + WorkflowNode, + WorkflowStepActionRequest, +) +from govoplan_workflow.backend.service import ( + WorkflowConflictError, + activate_definition, + create_definition, +) + + +def principal() -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + scopes=frozenset( + { + "workflow:definition:read", + "workflow:instance:read", + "workflow:instance:start", + "workflow:instance:transition", + "dataflow:pipeline:run", + } + ), + ), + account=object(), + user=object(), + ) + + +def runtime_graph() -> WorkflowGraph: + return WorkflowGraph( + nodes=[ + WorkflowNode( + id="start", + type="workflow.start.manual", + label="Start", + config={"input_schema_ref": ""}, + ), + WorkflowNode( + id="flow", + type="workflow.dataflow", + label="Prepare evidence", + config={ + "pipeline_ref": "pipeline:pipeline-1", + "revision": 3, + "environment": "development", + "row_limit": 250, + "publication_target_ref": "", + "warning_policy": "review", + "input_mapping": {}, + }, + ), + WorkflowNode( + id="complete", + type="workflow.end.completed", + label="Complete", + config={"output_mapping": {}}, + ), + WorkflowNode( + id="cancelled", + type="workflow.end.cancelled", + label="Rejected", + config={"reason": "Rejected during review"}, + ), + ], + edges=[ + WorkflowEdge( + id="start-flow", + source="start", + target="flow", + ), + WorkflowEdge( + id="flow-complete", + source="flow", + source_port="success", + target="complete", + ), + WorkflowEdge( + id="flow-warning", + source="flow", + source_port="warning", + target="complete", + ), + WorkflowEdge( + id="flow-review", + source="flow", + source_port="review_required", + target="complete", + ), + WorkflowEdge( + id="flow-failure", + source="flow", + source_port="failure", + target="cancelled", + ), + ], + ) + + +class FakeDataflowLifecycle: + def __init__(self) -> None: + self.runs: dict[str, DataflowRunDescriptor] = {} + self.requests = [] + self.cancelled: list[str] = [] + + def start_run(self, _session, _principal, *, request): + self.requests.append(request) + run_ref = f"run:{len(self.requests)}" + descriptor = DataflowRunDescriptor( + ref=run_ref, + pipeline_ref=request.pipeline_ref, + revision=request.revision, + status="queued", + definition_hash="definition-hash", + executor_version="test", + metadata={"progress_percent": 0, "progress_phase": "queued"}, + ) + self.runs[run_ref] = descriptor + return descriptor + + def get_run(self, _session, _principal, *, run_ref): + return self.runs.get(run_ref) + + def cancel_run(self, _session, _principal, *, run_ref): + descriptor = self.runs[run_ref] + descriptor = replace(descriptor, status="cancelled") + self.runs[run_ref] = descriptor + self.cancelled.append(run_ref) + return descriptor + + def finish( + self, + run_ref: str, + *, + diagnostics: list[dict[str, object]] | None = None, + ) -> None: + self.runs[run_ref] = replace( + self.runs[run_ref], + status="succeeded", + output_publication_ref="publication:1", + output_datasource_ref="datasource:1", + output_materialization_ref="materialization:1", + input_row_count=12, + output_row_count=10, + metadata={"diagnostics": diagnostics or []}, + ) + + def fail(self, run_ref: str) -> None: + self.runs[run_ref] = replace( + self.runs[run_ref], + status="failed", + error="Data quality gate failed.", + ) + + +class FakeAutomationProvider: + def __init__(self) -> None: + self.requests = [] + + def resolve_automation_principal(self, _session, *, request): + self.requests.append(request) + return AutomationPrincipalResolution( + allowed=True, + principal=principal(), + granted_scopes=request.grant_scopes, + provenance={"status": "rechecked"}, + ) + + +class Registry: + def __init__( + self, + dataflow: FakeDataflowLifecycle, + automation: FakeAutomationProvider | None = None, + ) -> None: + self.dataflow = dataflow + self.automation = automation + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_DATAFLOW_RUN_LIFECYCLE or ( + name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + and self.automation is not None + ) + + def capability(self, name: str): + if name == CAPABILITY_DATAFLOW_RUN_LIFECYCLE: + return self.dataflow + if ( + name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + and self.automation is not None + ): + return self.automation + raise KeyError(name) + + +class WorkflowInstanceServiceTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + WorkflowDefinition.__table__, + WorkflowDefinitionRevision.__table__, + WorkflowInstance.__table__, + WorkflowInstanceStep.__table__, + WorkflowInstanceEvent.__table__, + ], + ) + self.Session = sessionmaker(bind=self.engine) + self.session: Session = self.Session() + self.dataflow = FakeDataflowLifecycle() + self.registry = Registry(self.dataflow) + self.definition = create_definition( + self.session, + tenant_id="tenant-1", + actor_id="account-1", + payload=WorkflowDefinitionCreateRequest( + name="Monthly governed processing", + graph=runtime_graph(), + ), + ) + activate_definition( + self.session, + tenant_id="tenant-1", + definition_id=self.definition.id, + actor_id="account-1", + revision=1, + ) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all( + self.engine, + tables=[ + WorkflowInstanceEvent.__table__, + WorkflowInstanceStep.__table__, + WorkflowInstance.__table__, + WorkflowDefinitionRevision.__table__, + WorkflowDefinition.__table__, + ], + ) + self.engine.dispose() + + def _start(self, key: str = "request-1") -> WorkflowInstance: + instance, replayed = start_instance( + self.session, + tenant_id="tenant-1", + definition_id=self.definition.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key=key, + input={"case_id": "case-1"}, + correlation_id="correlation-1", + ), + ) + self.assertFalse(replayed) + return instance + + def test_start_pins_revision_and_replays_idempotently(self) -> None: + instance = self._start() + replayed, was_replayed = start_instance( + self.session, + tenant_id="tenant-1", + definition_id=self.definition.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key="request-1", + input={"case_id": "case-1"}, + correlation_id="correlation-1", + ), + ) + + self.assertTrue(was_replayed) + self.assertEqual(instance.id, replayed.id) + self.assertEqual("waiting", instance.status) + self.assertEqual(1, len(self.dataflow.requests)) + response = instance_response(self.session, instance) + self.assertEqual([1, 2], [step.sequence for step in response.steps]) + self.assertEqual("run:1", response.steps[-1].external_ref) + self.assertGreaterEqual(len(response.events), 4) + + with self.assertRaises(WorkflowConflictError): + start_instance( + self.session, + tenant_id="tenant-1", + definition_id=self.definition.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key="request-1", + input={"case_id": "another-case"}, + correlation_id="correlation-1", + ), + ) + + def test_reconcile_completes_with_stable_dataflow_output_refs(self) -> None: + instance = self._start() + self.dataflow.finish("run:1") + + changed = reconcile_instance( + self.session, + instance=instance, + principal=principal(), + registry=self.registry, + actor_id="account-1", + ) + response = instance_response(self.session, instance) + + self.assertTrue(changed) + self.assertEqual("completed", response.status) + flow_output = response.context["steps"]["flow"] + self.assertEqual("publication:1", flow_output["output_publication_ref"]) + self.assertEqual("datasource:1", flow_output["output_datasource_ref"]) + self.assertEqual( + "materialization:1", + flow_output["output_materialization_ref"], + ) + self.assertEqual("workflow.instance.completed", response.events[-1].kind) + + def test_warning_requires_review_and_approve_resumes(self) -> None: + instance = self._start() + self.dataflow.finish( + "run:1", + diagnostics=[ + { + "severity": "warning", + "code": "review.required", + "message": "Verify unmatched records.", + } + ], + ) + reconcile_instance( + self.session, + instance=instance, + principal=principal(), + registry=self.registry, + ) + step_id = str(instance.current_step_id) + + self.assertEqual( + "review_required", + instance_response(self.session, instance).steps[-1].handoff["state"], + ) + resolved = resolve_step( + self.session, + tenant_id="tenant-1", + instance_id=instance.id, + step_id=step_id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowStepActionRequest( + action="approve", + comment="Evidence verified.", + evidence=["publication:1"], + ), + ) + + self.assertEqual("completed", resolved.status) + + def test_failure_can_retry_and_reject_invalid_actions(self) -> None: + instance = self._start() + self.dataflow.fail("run:1") + reconcile_instance( + self.session, + instance=instance, + principal=principal(), + registry=self.registry, + ) + step_id = str(instance.current_step_id) + + with self.assertRaises(WorkflowConflictError): + resolve_step( + self.session, + tenant_id="tenant-1", + instance_id=instance.id, + step_id=step_id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowStepActionRequest(action="approve"), + ) + retried = resolve_step( + self.session, + tenant_id="tenant-1", + instance_id=instance.id, + step_id=step_id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowStepActionRequest(action="retry"), + ) + + self.assertEqual("waiting", retried.status) + self.assertEqual(2, len(self.dataflow.requests)) + self.assertEqual("run:2", retried.steps[-1].external_ref) + self.assertEqual("superseded", retried.steps[-2].status) + + def test_cancel_propagates_to_linked_dataflow(self) -> None: + instance = self._start() + + cancelled = cancel_instance( + self.session, + tenant_id="tenant-1", + instance_id=instance.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + ) + + self.assertEqual("cancelled", cancelled.status) + self.assertEqual(["run:1"], self.dataflow.cancelled) + + def test_worker_rechecks_authorization_before_reconciling(self) -> None: + instance = self._start() + self.session.commit() + self.dataflow.finish("run:1") + automation = FakeAutomationProvider() + worker = SqlWorkflowRuntimeWorker( + registry=Registry(self.dataflow, automation), + ) + + summary = worker.reconcile_pending(self.session) + + self.assertEqual(1, summary["advanced"]) + self.assertEqual("completed", instance.status) + self.assertEqual(1, len(automation.requests)) + self.assertEqual( + "rechecked", + instance.authorization_["last_resolution"]["status"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index b31c4a8..c78d941 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -5,8 +5,12 @@ import unittest from govoplan_workflow.backend.manifest import ( DEFINITION_READ_SCOPE, DEFINITION_WRITE_SCOPE, + INSTANCE_START_SCOPE, get_manifest, ) +from govoplan_core.core.workflows import ( + CAPABILITY_WORKFLOW_RUNTIME_WORKER, +) class WorkflowManifestTests(unittest.TestCase): @@ -26,6 +30,18 @@ class WorkflowManifestTests(unittest.TestCase): DEFINITION_WRITE_SCOPE, {item.scope for item in manifest.permissions}, ) + self.assertIn( + INSTANCE_START_SCOPE, + {item.scope for item in manifest.permissions}, + ) + self.assertIn( + "workflow.runtime_worker", + {item.name for item in manifest.provides_interfaces}, + ) + self.assertIn( + CAPABILITY_WORKFLOW_RUNTIME_WORKER, + manifest.capability_factories, + ) self.assertEqual( "@govoplan/workflow-webui", manifest.frontend.package_name if manifest.frontend else None, diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 83215a5..e17e56c 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -26,13 +26,16 @@ class WorkflowMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "c6d8f1a3e5b7", + "d8f2a5c7e1b4", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertEqual( { "workflow_definition_revisions", "workflow_definitions", + "workflow_instance_events", + "workflow_instance_steps", + "workflow_instances", }, { name diff --git a/webui/src/api/workflow.ts b/webui/src/api/workflow.ts index 4d8cbee..d288bb4 100644 --- a/webui/src/api/workflow.ts +++ b/webui/src/api/workflow.ts @@ -84,6 +84,75 @@ export type WorkflowGovernance = { automation_runtime_reason?: string | null; }; +export type WorkflowInstanceStatus = + | "running" + | "waiting" + | "completed" + | "failed" + | "cancelled"; + +export type WorkflowStepStatus = + | "running" + | "waiting" + | "completed" + | "failed" + | "cancelled" + | "superseded"; + +export type WorkflowInstanceStep = { + id: string; + sequence: number; + node_id: string; + node_type: string; + status: WorkflowStepStatus; + attempt: number; + input: Record; + output: Record; + handoff: Record; + external_ref?: string | null; + started_at?: string | null; + finished_at?: string | null; + error?: string | null; + completed_by?: string | null; + created_at: string; + updated_at: string; +}; + +export type WorkflowInstanceEvent = { + id: string; + sequence: number; + step_id?: string | null; + kind: string; + actor_id?: string | null; + payload: Record; + created_at: string; +}; + +export type WorkflowInstance = { + id: string; + definition_id: string; + definition_name: string; + definition_revision: number; + definition_hash: string; + status: WorkflowInstanceStatus; + idempotency_key: string; + correlation_id?: string | null; + current_step_id?: string | null; + input: Record; + context: Record; + output: Record; + started_at: string; + finished_at?: string | null; + cancellation_requested_at?: string | null; + error?: string | null; + created_by?: string | null; + created_at: string; + updated_at: string; + steps: WorkflowInstanceStep[]; + events: WorkflowInstanceEvent[]; + replayed: boolean; +}; + export type WorkflowDefinitionPayload = { name: string; description?: string | null; @@ -232,3 +301,89 @@ export function workflowScopeReferenceProvider( { scope_type: scopeType } ); } + +export async function listWorkflowInstances( + settings: ApiSettings, + definitionId?: string | null +): Promise { + const params = new URLSearchParams(); + if (definitionId) params.set("definition_id", definitionId); + const query = params.size ? `?${params.toString()}` : ""; + const response = await apiFetch<{ instances: WorkflowInstance[] }>( + settings, + `/api/v1/workflow/instances${query}` + ); + return response.instances; +} + +export function startWorkflowInstance( + settings: ApiSettings, + definitionId: string, + payload: { + idempotency_key: string; + input?: Record; + correlation_id?: string | null; + } +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/definitions/${encodeURIComponent(definitionId)}/instances`, + { + method: "POST", + body: JSON.stringify(payload) + } + ); +} + +export function getWorkflowInstance( + settings: ApiSettings, + instanceId: string +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/instances/${encodeURIComponent(instanceId)}` + ); +} + +export function reconcileWorkflowInstance( + settings: ApiSettings, + instanceId: string +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/instances/${encodeURIComponent(instanceId)}/reconcile`, + { method: "POST" } + ); +} + +export function resolveWorkflowStep( + settings: ApiSettings, + instanceId: string, + stepId: string, + payload: { + action: "complete" | "approve" | "changes" | "reject" | "resume" | "retry" | "cancel"; + output?: Record; + evidence?: string[]; + comment?: string | null; + } +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/instances/${encodeURIComponent(instanceId)}/steps/${encodeURIComponent(stepId)}/actions`, + { + method: "POST", + body: JSON.stringify(payload) + } + ); +} + +export function cancelWorkflowInstance( + settings: ApiSettings, + instanceId: string +): Promise { + return apiFetch( + settings, + `/api/v1/workflow/instances/${encodeURIComponent(instanceId)}/cancel`, + { method: "POST" } + ); +} diff --git a/webui/src/features/workflow/WorkflowPage.tsx b/webui/src/features/workflow/WorkflowPage.tsx index 12c3b4b..e875da8 100644 --- a/webui/src/features/workflow/WorkflowPage.tsx +++ b/webui/src/features/workflow/WorkflowPage.tsx @@ -10,6 +10,7 @@ import { CheckCircle2, CopyPlus, GitFork, + ListChecks, Plus, RefreshCw, RotateCcw, @@ -57,6 +58,7 @@ import WorkflowCanvas, { updateWorkflowGraphNode } from "./WorkflowCanvas"; import WorkflowInspector from "./WorkflowInspector"; +import WorkflowRunsDialog from "./WorkflowRunsDialog"; import { FALLBACK_WORKFLOW_LIBRARY, draftFromDefinition, @@ -103,6 +105,7 @@ export default function WorkflowPage({ const [deleteOpen, setDeleteOpen] = useState(false); const [definitionSettingsOpen, setDefinitionSettingsOpen] = useState(false); const [deriveOpen, setDeriveOpen] = useState(false); + const [runsOpen, setRunsOpen] = useState(false); const canWrite = hasScope(auth, "workflow:definition:write") || hasScope(auth, "workflow:instance:admin"); @@ -114,6 +117,18 @@ export default function WorkflowPage({ && canWrite && draft.governance?.actions.derive?.allowed ); + const canStart = Boolean( + draft?.id + && (hasScope(auth, "workflow:instance:start") + || hasScope(auth, "workflow:instance:admin")) + && draft.governance?.actions.start?.allowed !== false + ); + const canTransition = hasScope(auth, "workflow:instance:transition") + || hasScope(auth, "workflow:instance:admin"); + const selectedDefinition = useMemo( + () => definitions.find((item) => item.id === draft?.id) ?? null, + [definitions, draft?.id] + ); const dirty = Boolean(draft) && workflowFingerprint(draft) !== workflowFingerprint(savedDraft); const displayedGraph = historicalRevision?.graph ?? draft?.graph ?? null; @@ -531,6 +546,14 @@ export default function WorkflowPage({ + {draft.id ? ( + + ) : null} } @@ -748,6 +771,14 @@ export default function WorkflowPage({ setSuccess("Created a pinned scoped copy."); }} /> + setRunsOpen(false)} + /> ); } diff --git a/webui/src/features/workflow/WorkflowRunsDialog.tsx b/webui/src/features/workflow/WorkflowRunsDialog.tsx new file mode 100644 index 0000000..29c28e3 --- /dev/null +++ b/webui/src/features/workflow/WorkflowRunsDialog.tsx @@ -0,0 +1,516 @@ +import { + useCallback, + useEffect, + useMemo, + useState +} from "react"; +import { + ExternalLink, + Play, + RefreshCw, + RotateCcw, + XCircle +} from "lucide-react"; +import { + Button, + ConfirmDialog, + Dialog, + DismissibleAlert, + FormField, + IconButton, + LoadingFrame, + StatusBadge, + type ApiSettings +} from "@govoplan/core-webui"; +import { + cancelWorkflowInstance, + listWorkflowInstances, + reconcileWorkflowInstance, + resolveWorkflowStep, + startWorkflowInstance, + type WorkflowDefinition, + type WorkflowInstance, + type WorkflowInstanceStep +} from "../../api/workflow"; + +type WorkflowAction = + | "complete" + | "approve" + | "changes" + | "reject" + | "resume" + | "retry" + | "cancel"; + +export default function WorkflowRunsDialog({ + open, + settings, + definition, + canStart, + canTransition, + onClose +}: { + open: boolean; + settings: ApiSettings; + definition: WorkflowDefinition | null; + canStart: boolean; + canTransition: boolean; + onClose: () => void; +}) { + const [instances, setInstances] = useState([]); + const [selectedId, setSelectedId] = useState(null); + const [loading, setLoading] = useState(false); + const [working, setWorking] = useState(false); + const [error, setError] = useState(""); + const [comment, setComment] = useState(""); + const [evidence, setEvidence] = useState(""); + const [cancelOpen, setCancelOpen] = useState(false); + + const selected = useMemo( + () => instances.find((item) => item.id === selectedId) ?? instances[0] ?? null, + [instances, selectedId] + ); + const currentStep = useMemo( + () => currentInstanceStep(selected), + [selected] + ); + const allowedActions = useMemo( + () => handoffActions(currentStep), + [currentStep] + ); + + const mergeInstance = useCallback((instance: WorkflowInstance) => { + setInstances((current) => [ + instance, + ...current.filter((item) => item.id !== instance.id) + ]); + setSelectedId(instance.id); + }, []); + + const load = useCallback(async () => { + if (!open || !definition?.id) return; + setLoading(true); + setError(""); + try { + const items = await listWorkflowInstances(settings, definition.id); + setInstances(items); + setSelectedId((current) => ( + items.some((item) => item.id === current) + ? current + : items[0]?.id ?? null + )); + } catch (loadError) { + setError(errorMessage(loadError)); + } finally { + setLoading(false); + } + }, [definition?.id, open, settings]); + + useEffect(() => { + if (!open) return; + setComment(""); + setEvidence(""); + void load(); + }, [load, open]); + + useEffect(() => { + if ( + !open + || !canTransition + || !selected + || !currentStep + || currentStep.node_type !== "workflow.dataflow" + || !["queued", "retrying", "running"].includes( + String(currentStep.handoff.state ?? "") + ) + ) { + return; + } + let stopped = false; + const poll = window.setInterval(() => { + void reconcileWorkflowInstance(settings, selected.id) + .then((instance) => { + if (!stopped) mergeInstance(instance); + }) + .catch((pollError) => { + if (!stopped) setError(errorMessage(pollError)); + }); + }, 2500); + return () => { + stopped = true; + window.clearInterval(poll); + }; + }, [ + canTransition, + currentStep, + mergeInstance, + open, + selected, + settings + ]); + + const start = async () => { + if (!definition?.id) return; + setWorking(true); + setError(""); + try { + const instance = await startWorkflowInstance(settings, definition.id, { + idempotency_key: crypto.randomUUID(), + input: {} + }); + mergeInstance(instance); + } catch (startError) { + setError(errorMessage(startError)); + } finally { + setWorking(false); + } + }; + + const refreshSelected = async () => { + if (!selected) { + await load(); + return; + } + setWorking(true); + setError(""); + try { + const instance = canTransition + ? await reconcileWorkflowInstance(settings, selected.id) + : (await listWorkflowInstances(settings, definition?.id)) + .find((item) => item.id === selected.id); + if (instance) mergeInstance(instance); + else await load(); + } catch (refreshError) { + setError(errorMessage(refreshError)); + } finally { + setWorking(false); + } + }; + + const performAction = async (action: WorkflowAction) => { + if (!selected || !currentStep) return; + setWorking(true); + setError(""); + try { + const instance = await resolveWorkflowStep( + settings, + selected.id, + currentStep.id, + { + action, + comment: comment.trim() || null, + evidence: evidence + .split("\n") + .map((item) => item.trim()) + .filter(Boolean) + } + ); + mergeInstance(instance); + setComment(""); + setEvidence(""); + } catch (actionError) { + setError(errorMessage(actionError)); + } finally { + setWorking(false); + } + }; + + const cancel = async () => { + if (!selected) return; + setWorking(true); + setError(""); + try { + mergeInstance(await cancelWorkflowInstance(settings, selected.id)); + setCancelOpen(false); + } catch (cancelError) { + setError(errorMessage(cancelError)); + } finally { + setWorking(false); + } + }; + + const actionUrl = typeof currentStep?.handoff.action_url === "string" + ? currentStep.handoff.action_url + : ""; + + return ( + <> + Close} + > +
+ + Workflow instances + Revision-pinned runs and human handoffs + + + } + variant="ghost" + onClick={() => void refreshSelected()} + disabled={loading || working} + /> + + +
+ {error ? ( + + {error} + + ) : null} + +
+
+ {instances.map((instance) => ( + + ))} + {!instances.length ? ( +
No runs yet
+ ) : null} +
+
+ {selected ? ( + <> +
+ + {selected.definition_name} + + Revision {selected.definition_revision} · {selected.definition_hash.slice(0, 12)} + + + + + {["running", "waiting"].includes(selected.status) ? ( + } + variant="danger" + onClick={() => setCancelOpen(true)} + disabled={!canTransition || working} + /> + ) : null} + +
+ {selected.error ? ( + + {selected.error} + + ) : null} + {currentStep ? ( +
+
+ + + {String( + currentStep.handoff.title + ?? currentStep.handoff.kind + ?? currentStep.node_type + )} + + + Step {currentStep.sequence} · attempt {currentStep.attempt} + + + +
+ {typeof currentStep.handoff.message === "string" ? ( +

{currentStep.handoff.message}

+ ) : null} + {typeof currentStep.handoff.instructions === "string" + && currentStep.handoff.instructions ? ( +

{currentStep.handoff.instructions}

+ ) : null} + {actionUrl ? ( + + Open linked Dataflow result + + ) : null} + {allowedActions.some((action) => action !== "cancel") ? ( +
+ +