GovOPlaN/govoplan-workflow-engine
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Add durable workflow triggers and waits

Browse Source
This commit is contained in:
zemion
2026-08-01 20:57:26 +02:00
parent 55f98d1b65
commit a1cea1d162
12 changed files with 2213 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_workflow_engine/backend/instance_service.py
+252 -194
View File
@@ -29,6 +29,13 @@ from govoplan_core.core.notifications import (
NotificationDispatchRequest,
notification_dispatch_provider,
)
from govoplan_core.core.events import (
EventActorRef,
EventObjectRef,
EventTenantRef,
PlatformEvent,
emit_platform_event,
)
from govoplan_core.db.base import utcnow
from govoplan_workflow_engine.backend.db.models import (
WorkflowDefinition,
@@ -89,9 +96,7 @@ def list_instances(
.limit(max(1, min(int(limit), 200)))
)
if definition_id:
statement = statement.where(
WorkflowInstance.definition_id == definition_id
)
statement = statement.where(WorkflowInstance.definition_id == definition_id)
return list(session.scalars(statement))
@@ -160,9 +165,7 @@ def start_instance(
)
normalized_origin = _normalize_start_origin(start_origin)
if normalized_origin != "user" and not definition.allow_automation:
raise WorkflowConflictError(
"This Workflow does not allow automated starts."
)
raise WorkflowConflictError("This Workflow does not allow automated starts.")
if revision.execution_mode == "guided" and normalized_origin != "user":
raise WorkflowConflictError(
"Guided workflows must be started by a user; use hybrid mode "
@@ -189,8 +192,7 @@ def start_instance(
or existing.start_origin != normalized_origin
):
raise WorkflowConflictError(
"The Workflow idempotency key was already used with "
"different input."
"The Workflow idempotency key was already used with different input."
)
return get_instance(
session,
@@ -357,9 +359,7 @@ def reconcile_instance(
step.handoff = {
**dict(step.handoff),
"state": descriptor.status,
"progress_percent": int(
descriptor.metadata.get("progress_percent") or 0
),
"progress_percent": int(descriptor.metadata.get("progress_percent") or 0),
"progress_phase": str(
descriptor.metadata.get("progress_phase") or descriptor.status
),
@@ -434,14 +434,11 @@ def resolve_step(
instance.definition_revision_id,
)
if revision is None:
raise WorkflowConflictError(
"Pinned Workflow revision no longer exists."
)
raise WorkflowConflictError("Pinned Workflow revision no longer exists.")
graph = _runtime_graph(revision)
node = _node(graph, step.node_id)
allowed_actions = {
str(action)
for action in step.handoff.get("allowed_actions") or ()
str(action) for action in step.handoff.get("allowed_actions") or ()
}
if payload.action not in allowed_actions:
raise WorkflowConflictError(
@@ -525,6 +522,10 @@ def resolve_step(
"comment": payload.comment,
"evidence": list(payload.evidence),
}
if step.node_type == "workflow.wait":
from govoplan_workflow_engine.backend.triggers import resolve_wait_state
resolve_wait_state(session, step_id=step.id, status="resumed")
next_node_id = _complete_step(
session,
instance=instance,
@@ -562,28 +563,33 @@ def cancel_instance(
for_update=True,
)
if instance.status in {"completed", "failed", "cancelled"}:
raise WorkflowConflictError(
f"Workflow instance is already {instance.status}."
)
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,
)
if step is not None:
if step.external_ref:
provider = dataflow_run_lifecycle(registry)
if provider is not None:
try:
provider.cancel_run(
session,
principal,
run_ref=step.external_ref,
)
except ValueError as exc:
logger.info(
"Linked Dataflow run could not be cancelled for "
"Workflow instance %s: %s",
instance.id,
exc,
)
if step.node_type == "workflow.wait":
from govoplan_workflow_engine.backend.triggers import (
resolve_wait_state,
)
resolve_wait_state(session, step_id=step.id, status="cancelled")
step.status = "cancelled"
step.finished_at = now
step.completed_by = actor_id
@@ -805,7 +811,6 @@ def _drive_instance(
if node.type in {
"workflow.activity",
"workflow.review",
"workflow.wait",
}:
_set_human_handoff(
session,
@@ -815,6 +820,34 @@ def _drive_instance(
registry=registry,
)
return
if node.type == "workflow.wait":
from govoplan_workflow_engine.backend.triggers import (
register_wait_state,
)
wait_state = register_wait_state(
session,
instance=instance,
step=step,
node=node,
)
if wait_state is None:
_set_human_handoff(
session,
instance=instance,
step=step,
node=node,
registry=registry,
)
else:
_set_automated_wait(
session,
instance=instance,
step=step,
node=node,
wait_state=wait_state,
)
return
if node.type == "workflow.end.completed":
_complete_step(
session,
@@ -1096,9 +1129,7 @@ def _execute_capability_step(
"manual_required",
"compensation_required",
}
announced_effects = {
item.effect_key for item in provider.effect_definitions()
}
announced_effects = {item.effect_key for item in provider.effect_definitions()}
unknown_effects = sorted(
{
effect.effect_key
@@ -1210,25 +1241,19 @@ def _capability_action_definition(
f"Action capability {capability_name!r} is not available."
)
definitions = [
item
for item in provider.action_definitions()
if item.action_key == action_key
item for item in provider.action_definitions() if item.action_key == action_key
]
if len(definitions) != 1:
raise WorkflowConflictError(
f"Action {action_key!r} is not uniquely announced by "
f"{capability_name!r}."
f"Action {action_key!r} is not uniquely announced by {capability_name!r}."
)
definition = definitions[0]
missing_scopes = [
scope
for scope in definition.required_scopes
if not has_scope(principal, scope)
scope for scope in definition.required_scopes if not has_scope(principal, scope)
]
if missing_scopes:
raise WorkflowConflictError(
"Module action requires scopes: "
+ ", ".join(sorted(missing_scopes))
"Module action requires scopes: " + ", ".join(sorted(missing_scopes))
)
missing_capabilities = [
capability
@@ -1244,12 +1269,8 @@ def _capability_action_definition(
"Module action requires capabilities: "
+ ", ".join(sorted(missing_capabilities))
)
effect_keys = {
item.effect_key for item in provider.effect_definitions()
}
missing_effects = sorted(
set(definition.expected_effect_keys) - effect_keys
)
effect_keys = {item.effect_key for item in provider.effect_definitions()}
missing_effects = sorted(set(definition.expected_effect_keys) - effect_keys)
if missing_effects:
raise WorkflowConflictError(
"Action provider does not define its expected effects: "
@@ -1265,9 +1286,7 @@ def _mapped_action_input(
if raw_mapping is None or raw_mapping == "":
return dict(context)
if not isinstance(raw_mapping, Mapping):
raise WorkflowConflictError(
"Module-action input mapping must be an object."
)
raise WorkflowConflictError("Module-action input mapping must be an object.")
return {
str(key): _resolve_action_value(value, context, depth=0)
for key, value in raw_mapping.items()
@@ -1282,9 +1301,7 @@ def _resolve_action_value(
depth: int,
) -> object:
if depth > 10:
raise WorkflowConflictError(
"Module-action input mapping is nested too deeply."
)
raise WorkflowConflictError("Module-action input mapping is nested too deeply.")
if isinstance(value, str) and value.startswith("$"):
path = value[1:].lstrip(".")
current: object = context
@@ -1307,10 +1324,7 @@ def _resolve_action_value(
for key, nested in value.items()
}
if isinstance(value, list):
return [
_resolve_action_value(item, context, depth=depth + 1)
for item in value
]
return [_resolve_action_value(item, context, depth=depth + 1) for item in value]
return value
@@ -1322,9 +1336,7 @@ def _action_idempotency_key(
action_key: str,
context: Mapping[str, object],
) -> str:
expression = str(
node.config.get("idempotency_key") or "workflow-step"
).strip()
expression = str(node.config.get("idempotency_key") or "workflow-step").strip()
if expression == "workflow-step":
return step.idempotency_key
resolved = _resolve_action_value(expression, context, depth=0)
@@ -1345,8 +1357,7 @@ def _action_preview_payload(preview: object) -> dict[str, object]:
"preview_ref": getattr(preview, "preview_ref", None),
"blockers": list(getattr(preview, "blockers", ()) or ()),
"policy_provenance": [
dict(item)
for item in getattr(preview, "policy_provenance", ()) or ()
dict(item) for item in getattr(preview, "policy_provenance", ()) or ()
],
"effects": [
{
@@ -1380,9 +1391,7 @@ def _action_result_payload(
],
"error": result.error,
"retry_after": (
result.retry_after.isoformat()
if result.retry_after is not None
else None
result.retry_after.isoformat() if result.retry_after is not None else None
),
"manual_instructions": result.manual_instructions,
"compensation_action_key": result.compensation_action_key,
@@ -1403,9 +1412,7 @@ def _set_action_handoff(
details: Mapping[str, object] | None = None,
) -> None:
allowed_actions = (
["cancel"]
if state in {"pending", "running"}
else ["retry", "reject", "cancel"]
["cancel"] if state in {"pending", "running"} else ["retry", "reject", "cancel"]
)
previous = dict(step.handoff)
step.status = "waiting"
@@ -1422,10 +1429,7 @@ def _set_action_handoff(
}
instance.status = "waiting"
instance.error = step.error
if (
previous.get("state") != state
or previous.get("message") != message
):
if previous.get("state") != state or previous.get("message") != message:
_record_event(
session,
instance,
@@ -1497,9 +1501,7 @@ def _start_dataflow_step(
message="Dataflow steps require a pipeline and pinned revision.",
)
return
target_ref = str(
node.config.get("publication_target_ref") or ""
).strip()
target_ref = str(node.config.get("publication_target_ref") or "").strip()
try:
run = provider.start_run(
session,
@@ -1509,13 +1511,9 @@ def _start_dataflow_step(
revision=revision,
idempotency_key=step.idempotency_key,
row_limit=row_limit,
environment=str(
node.config.get("environment") or "development"
),
environment=str(node.config.get("environment") or "development"),
publication=(
DataflowPublicationTarget(
target_datasource_ref=target_ref
)
DataflowPublicationTarget(target_datasource_ref=target_ref)
if target_ref
else None
),
@@ -1525,9 +1523,7 @@ def _start_dataflow_step(
causation_id=f"workflow-step:{step.id}",
requested_by=instance.created_by,
metadata={
"workflow_instance_ref": (
f"workflow-instance:{instance.id}"
),
"workflow_instance_ref": (f"workflow-instance:{instance.id}"),
"workflow_step_ref": f"workflow-step:{step.id}",
},
),
@@ -1557,12 +1553,8 @@ def _start_dataflow_step(
"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
),
"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(
@@ -1592,11 +1584,11 @@ def _handle_dataflow_success(
warnings = [
dict(item)
for item in items
if isinstance(item, Mapping)
and str(item.get("severity") or "") == "warning"
if isinstance(item, Mapping) and str(item.get("severity") or "") == "warning"
]
explicit_review = any(
str(item.get("code") or "") in {
str(item.get("code") or "")
in {
"review.required",
"reconciliation.review_required",
}
@@ -1606,8 +1598,7 @@ def _handle_dataflow_success(
output = _dataflow_output(descriptor)
step.output_ = output
if explicit_review or (
warnings
and str(node.config.get("warning_policy") or "review") == "review"
warnings and str(node.config.get("warning_policy") or "review") == "review"
):
step.status = "waiting"
step.handoff = {
@@ -1626,9 +1617,7 @@ def _handle_dataflow_success(
"retry",
"cancel",
],
"suggested_port": (
"review_required" if explicit_review else "warning"
),
"suggested_port": ("review_required" if explicit_review else "warning"),
"warnings": warnings,
"output": output,
}
@@ -1729,11 +1718,8 @@ def _set_human_handoff(
"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 []
),
"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"
@@ -1754,6 +1740,38 @@ def _set_human_handoff(
)
def _set_automated_wait(
session: Session,
*,
instance: WorkflowInstance,
step: WorkflowInstanceStep,
node: WorkflowNode,
wait_state: object,
) -> None:
mode = str(getattr(wait_state, "mode"))
due_at = getattr(wait_state, "due_at")
event_type = getattr(wait_state, "event_type")
step.status = "waiting"
step.handoff = {
"kind": "event_wait" if mode == "event" else "timer",
"state": "waiting",
"title": str(node.config.get("title") or node.label or "Wait"),
"mode": mode,
"due_at": due_at.isoformat() if due_at else None,
"event_type": event_type,
"allowed_actions": ["cancel"],
}
instance.status = "waiting"
_record_event(
session,
instance,
step=step,
kind="workflow.wait.registered",
actor_id=instance.created_by,
payload=dict(step.handoff),
)
def _set_dependency_handoff(
session: Session,
*,
@@ -1913,25 +1931,31 @@ def _new_step(
instance: WorkflowInstance,
node: WorkflowNode,
) -> WorkflowInstanceStep:
sequence = int(
session.scalar(
select(func.max(WorkflowInstanceStep.sequence)).where(
WorkflowInstanceStep.instance_id == instance.id
sequence = (
int(
session.scalar(
select(func.max(WorkflowInstanceStep.sequence)).where(
WorkflowInstanceStep.instance_id == instance.id
)
)
or 0
)
or 0
) + 1
attempt = int(
session.scalar(
select(func.count())
.select_from(WorkflowInstanceStep)
.where(
WorkflowInstanceStep.instance_id == instance.id,
WorkflowInstanceStep.node_id == node.id,
+ 1
)
attempt = (
int(
session.scalar(
select(func.count())
.select_from(WorkflowInstanceStep)
.where(
WorkflowInstanceStep.instance_id == instance.id,
WorkflowInstanceStep.node_id == node.id,
)
)
or 0
)
or 0
) + 1
+ 1
)
step = WorkflowInstanceStep(
tenant_id=instance.tenant_id,
instance=instance,
@@ -1940,9 +1964,7 @@ def _new_step(
node_type=node.type,
status="running",
attempt=attempt,
idempotency_key=(
f"workflow:{instance.id}:node:{node.id}:attempt:{attempt}"
),
idempotency_key=(f"workflow:{instance.id}:node:{node.id}:attempt:{attempt}"),
input_=dict(instance.context_),
output_={},
handoff={},
@@ -1962,14 +1984,17 @@ def _record_event(
payload: Mapping[str, object],
step: WorkflowInstanceStep | None = None,
) -> None:
sequence = int(
session.scalar(
select(func.max(WorkflowInstanceEvent.sequence)).where(
WorkflowInstanceEvent.instance_id == instance.id
sequence = (
int(
session.scalar(
select(func.max(WorkflowInstanceEvent.sequence)).where(
WorkflowInstanceEvent.instance_id == instance.id
)
)
or 0
)
or 0
) + 1
+ 1
)
event = WorkflowInstanceEvent(
tenant_id=instance.tenant_id,
instance=instance,
@@ -1982,6 +2007,46 @@ def _record_event(
)
session.add(event)
session.flush()
if kind.startswith("workflow.instance."):
from govoplan_workflow_engine.backend.runtime import get_registry
emit_platform_event(
session,
PlatformEvent(
type=kind,
module_id="workflow_engine",
event_id=event.id,
occurred_at=event.created_at,
correlation_id=instance.correlation_id,
causation_id=(
str(payload.get("event_id")) if payload.get("event_id") else None
),
actor=(
EventActorRef(type="account", id=actor_id)
if actor_id
else EventActorRef(type="system_actor")
),
tenant=EventTenantRef(id=instance.tenant_id),
subject=EventObjectRef(
type="workflow_instance",
id=instance.id,
),
resource=EventObjectRef(
type="workflow_definition",
id=instance.definition_id,
),
classification="internal",
payload={
"instance_id": instance.id,
"definition_id": instance.definition_id,
"definition_revision_id": instance.definition_revision_id,
"step_id": step.id if step else None,
"status": instance.status,
"start_origin": instance.start_origin,
},
),
registry=get_registry(),
)
def _current_step(
@@ -1998,15 +2063,12 @@ def _start_node(graph: WorkflowGraph, *, kind: str) -> WorkflowNode:
node = next((item for item in graph.nodes if item.type == expected), None)
if node is None:
starts = [
item for item in graph.nodes
if item.type.startswith("workflow.start.")
item for item in graph.nodes if item.type.startswith("workflow.start.")
]
if len(starts) == 1:
node = starts[0]
if node is None:
raise WorkflowConflictError(
f"Workflow has no {kind} start node."
)
raise WorkflowConflictError(f"Workflow has no {kind} start node.")
return node
@@ -2024,9 +2086,7 @@ def _normalize_start_origin(value: str) -> str:
"backfill",
}
if normalized not in allowed:
raise WorkflowConflictError(
f"Unsupported Workflow start origin {value!r}."
)
raise WorkflowConflictError(f"Unsupported Workflow start origin {value!r}.")
return normalized
@@ -2048,17 +2108,10 @@ def _instance_view_context(
instance: WorkflowInstance,
revision: WorkflowDefinitionRevision,
) -> WorkflowViewContextResponse | None:
if (
not revision.view_id
or instance.status not in {"running", "waiting"}
):
if not revision.view_id or instance.status not in {"running", "waiting"}:
return None
step = next(
(
item
for item in instance.steps
if item.id == instance.current_step_id
),
(item for item in instance.steps if item.id == instance.current_step_id),
None,
)
node = None
@@ -2089,17 +2142,13 @@ def _instance_view_context(
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."
)
raise WorkflowConflictError(f"Workflow node {node_id!r} no longer exists.")
return node
def _runtime_graph(revision: WorkflowDefinitionRevision) -> WorkflowGraph:
try:
return materialize_runtime_graph(
WorkflowGraph.model_validate(revision.graph)
)
return materialize_runtime_graph(WorkflowGraph.model_validate(revision.graph))
except BpmnGraphError as exc:
raise WorkflowConflictError(str(exc)) from exc
@@ -2157,9 +2206,7 @@ def _dataflow_output(
"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 []
),
"diagnostics": list(descriptor.metadata.get("diagnostics") or []),
}
@@ -2199,6 +2246,32 @@ def _authorization_payload(
registry: object | None,
) -> dict[str, object]:
principal_ref = principal.to_platform_principal()
scopes = required_instance_scopes(
graph,
principal=principal,
registry=registry,
)
return {
"contract_version": "1",
"subject_kind": (
"service_account" if principal_ref.service_account_id else "delegated_user"
),
"account_id": principal_ref.account_id,
"membership_id": principal_ref.membership_id,
"service_account_id": principal_ref.service_account_id,
"grant_scopes": list(scopes),
"authorization_ref": None,
}
def required_instance_scopes(
graph: WorkflowGraph,
*,
principal: ApiPrincipal,
registry: object | None,
) -> tuple[str, ...]:
"""Return scopes pinned into an instance or trigger authorization artifact."""
scopes = {INSTANCE_START_SCOPE}
if any(node.type == "workflow.dataflow" for node in graph.nodes):
scopes.add(DATAFLOW_RUN_SCOPE)
@@ -2211,19 +2284,7 @@ def _authorization_payload(
registry=registry,
)
scopes.update(definition.required_scopes)
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,
}
return tuple(sorted(scopes))
def _resolve_instance_principal(
@@ -2239,25 +2300,18 @@ def _resolve_instance_principal(
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 ()
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}"
),
"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 ""
),
service_account_id=str(value.get("service_account_id") or ""),
**common,
)
else:
@@ -2287,8 +2341,7 @@ def _resolve_instance_principal(
}
return (
resolution.principal
if resolution.allowed
and isinstance(resolution.principal, ApiPrincipal)
if resolution.allowed and isinstance(resolution.principal, ApiPrincipal)
else None
)
@@ -2302,9 +2355,7 @@ def _notify_handoff(
subject: str,
) -> None:
provider = notification_dispatch_provider(registry)
account_id = str(
instance.authorization_.get("account_id") or ""
).strip()
account_id = str(instance.authorization_.get("account_id") or "").strip()
if provider is None or not account_id:
return
try:
@@ -2320,9 +2371,7 @@ def _notify_handoff(
recipient_id=account_id,
subject=subject,
action_url=(
"/workflow?"
f"definition={instance.definition_id}"
f"&run={instance.id}"
f"/workflow?definition={instance.definition_id}&run={instance.id}"
),
payload={
"instance_id": instance.id,
@@ -2351,7 +2400,6 @@ class SqlWorkflowRuntimeWorker:
now: datetime | None = None,
limit: int = 50,
) -> Mapping[str, object]:
del now
if not isinstance(session, Session):
raise TypeError("Workflow reconciliation requires a Session.")
standards: Mapping[str, object] | None = None
@@ -2370,8 +2418,17 @@ class SqlWorkflowRuntimeWorker:
registry=self._registry,
limit=limit,
)
from govoplan_workflow_engine.backend.triggers import dispatch_due_work
triggers = dispatch_due_work(
session,
registry=self._registry,
now=now,
limit=limit,
)
return {
**runtime,
"triggers": triggers,
**({"standards": standards} if standards is not None else {}),
}
@@ -2384,6 +2441,7 @@ __all__ = [
"list_instances",
"reconcile_instance",
"reconcile_pending_instances",
"required_instance_scopes",
"resolve_step",
"start_instance",
]