refactor: retain workflow as optional editor

This commit is contained in:
2026-07-31 16:59:02 +02:00
parent 484ac43352
commit 8fd8753012
33 changed files with 450 additions and 11483 deletions
+1 -1
View File
@@ -1,3 +1,3 @@
"""GovOPlaN Workflow module."""
"""Optional GovOPlaN Workflow editor and compatibility package."""
__version__ = "0.1.14"
+1 -1
View File
@@ -1 +1 @@
"""Workflow backend."""
"""Compatibility facades for the extracted Workflow Engine backend."""
+2 -343
View File
@@ -1,344 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
from collections import Counter
from dataclasses import dataclass
from typing import Literal
from xml.etree.ElementTree import Element, ParseError
from defusedxml.ElementTree import fromstring
from defusedxml.common import DefusedXmlException
BPMN_MODEL_NAMESPACE = "http://www.omg.org/spec/BPMN/20100524/MODEL"
BPMN_DI_NAMESPACE = "http://www.omg.org/spec/BPMN/20100524/DI"
OMG_DI_NAMESPACE = "http://www.omg.org/spec/DD/20100524/DI"
OMG_DC_NAMESPACE = "http://www.omg.org/spec/DD/20100524/DC"
MAX_BPMN_XML_BYTES = 1_048_576
MAX_BPMN_ELEMENTS = 20_000
SupportLevel = Literal[
"interchange_only",
"native_mapping",
"native_execution",
]
NATIVE_EXECUTION_ELEMENTS = frozenset(
{
"startEvent",
"endEvent",
"task",
"manualTask",
"userTask",
"serviceTask",
"sendTask",
"receiveTask",
"intermediateCatchEvent",
"sequenceFlow",
}
)
NATIVE_MAPPING_ELEMENTS = NATIVE_EXECUTION_ELEMENTS | frozenset(
{
"definitions",
"process",
"collaboration",
"documentation",
"extensionElements",
"incoming",
"outgoing",
"conditionExpression",
"scriptTask",
"serviceTask",
"businessRuleTask",
"receiveTask",
"sendTask",
"callActivity",
"subProcess",
"transaction",
"adHocSubProcess",
"exclusiveGateway",
"parallelGateway",
"inclusiveGateway",
"eventBasedGateway",
"complexGateway",
"boundaryEvent",
"eventSubProcess",
"intermediateCatchEvent",
"intermediateThrowEvent",
"dataObject",
"dataObjectReference",
"dataStoreReference",
"messageFlow",
"participant",
"lane",
"laneSet",
"textAnnotation",
"association",
"group",
"dataInputAssociation",
"dataOutputAssociation",
"choreography",
"choreographyTask",
"callChoreography",
"subChoreography",
"conversation",
"callConversation",
"subConversation",
"conversationLink",
}
)
@dataclass(frozen=True, slots=True)
class BpmnDiagnostic:
severity: Literal["error", "warning", "info"]
code: str
message: str
element_id: str | None = None
@dataclass(frozen=True, slots=True)
class BpmnElementInventoryItem:
element_type: str
element_id: str | None
name: str | None
parent_type: str | None
parent_id: str | None
support_level: SupportLevel
@dataclass(frozen=True, slots=True)
class BpmnInspection:
definitions_id: str | None
target_namespace: str | None
process_count: int
executable_process_count: int
collaboration_count: int
choreography_count: int
element_counts: dict[str, int]
support_counts: dict[str, int]
elements: tuple[BpmnElementInventoryItem, ...]
diagnostics: tuple[BpmnDiagnostic, ...]
@property
def valid_xml(self) -> bool:
return not any(item.severity == "error" for item in self.diagnostics)
class BpmnInspectionError(ValueError):
pass
def bpmn_support_level(element_type: str) -> SupportLevel:
if element_type in NATIVE_EXECUTION_ELEMENTS:
return "native_execution"
if element_type in NATIVE_MAPPING_ELEMENTS:
return "native_mapping"
return "interchange_only"
def parse_bpmn_xml(xml: str) -> Element:
encoded = xml.encode("utf-8")
if not encoded:
raise BpmnInspectionError("BPMN XML is empty")
if len(encoded) > MAX_BPMN_XML_BYTES:
raise BpmnInspectionError(
f"BPMN XML exceeds the {MAX_BPMN_XML_BYTES}-byte inspection limit"
)
try:
root = fromstring(encoded)
except (DefusedXmlException, ParseError, ValueError) as exc:
raise BpmnInspectionError(f"BPMN XML is not safe and well formed: {exc}") from exc
namespace, local_name = _qualified_name(root.tag)
if namespace != BPMN_MODEL_NAMESPACE or local_name != "definitions":
raise BpmnInspectionError(
"BPMN document root must be bpmn:definitions in the BPMN 2.0 model namespace"
)
if sum(1 for _item in root.iter()) > MAX_BPMN_ELEMENTS:
raise BpmnInspectionError(
f"BPMN document exceeds the {MAX_BPMN_ELEMENTS}-element inspection limit"
)
return root
def inspect_bpmn_xml(xml: str) -> BpmnInspection:
root = parse_bpmn_xml(xml)
diagnostics: list[BpmnDiagnostic] = []
elements: list[BpmnElementInventoryItem] = []
ids: dict[str, str] = {}
references: list[tuple[str, str | None, str]] = []
process_count = 0
executable_process_count = 0
collaboration_count = 0
choreography_count = 0
stack = [(root, None, None)]
visited = 0
while stack:
element, parent_type, parent_id = stack.pop()
visited += 1
element_namespace, element_type = _qualified_name(element.tag)
element_id = _bounded_attribute(element.attrib.get("id"), 255)
if element_namespace == BPMN_MODEL_NAMESPACE:
name = _bounded_attribute(element.attrib.get("name"), 300)
support = bpmn_support_level(element_type)
elements.append(
BpmnElementInventoryItem(
element_type=element_type,
element_id=element_id,
name=name,
parent_type=parent_type,
parent_id=parent_id,
support_level=support,
)
)
if element_id:
previous = ids.get(element_id)
if previous is not None:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="duplicate_bpmn_id",
message=(
f"BPMN id {element_id!r} is used by both "
f"{previous} and {element_type}"
),
element_id=element_id,
)
)
else:
ids[element_id] = element_type
if element_type == "process":
process_count += 1
if element.attrib.get("isExecutable", "").strip().lower() == "true":
executable_process_count += 1
elif element_type == "collaboration":
collaboration_count += 1
elif element_type in {"choreography", "globalChoreographyTask"}:
choreography_count += 1
_collect_references(element_type, element_id, element.attrib, references)
next_parent_type = element_type
next_parent_id = element_id
else:
next_parent_type = parent_type
next_parent_id = parent_id
for child in reversed(list(element)):
stack.append((child, next_parent_type, next_parent_id))
if process_count == 0 and collaboration_count == 0 and choreography_count == 0:
diagnostics.append(
BpmnDiagnostic(
severity="warning",
code="no_bpmn_process_or_collaboration",
message="BPMN definitions contain no process, collaboration, or choreography",
)
)
for reference, element_id, field in references:
if reference not in ids:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="dangling_bpmn_reference",
message=(
f"{field} references unknown BPMN id {reference!r}"
),
element_id=element_id,
)
)
unsupported = [
item
for item in elements
if item.support_level == "interchange_only"
and item.element_type not in {"definitions", "process"}
]
if unsupported:
diagnostics.append(
BpmnDiagnostic(
severity="info",
code="interchange_only_elements",
message=(
f"{len(unsupported)} BPMN element(s) can be inventoried and "
"round-tripped by a BPMN modeler but are not executable by "
"the native GovOPlaN workflow runtime"
),
)
)
element_counts = dict(
sorted(Counter(item.element_type for item in elements).items())
)
support_counts = dict(
sorted(Counter(item.support_level for item in elements).items())
)
return BpmnInspection(
definitions_id=_bounded_attribute(root.attrib.get("id"), 255),
target_namespace=_bounded_attribute(
root.attrib.get("targetNamespace"),
1000,
),
process_count=process_count,
executable_process_count=executable_process_count,
collaboration_count=collaboration_count,
choreography_count=choreography_count,
element_counts=element_counts,
support_counts=support_counts,
elements=tuple(elements),
diagnostics=tuple(diagnostics),
)
def _collect_references(
element_type: str,
element_id: str | None,
attributes: dict[str, str],
references: list[tuple[str, str | None, str]],
) -> None:
fields_by_element = {
"sequenceFlow": ("sourceRef", "targetRef"),
"messageFlow": ("sourceRef", "targetRef", "messageRef"),
"association": ("sourceRef", "targetRef"),
"participant": ("processRef",),
"lane": ("partitionElementRef",),
"boundaryEvent": ("attachedToRef",),
"dataInputAssociation": ("sourceRef", "targetRef"),
"dataOutputAssociation": ("sourceRef", "targetRef"),
"dataObjectReference": ("dataObjectRef",),
"dataStoreReference": ("dataStoreRef",),
"messageEventDefinition": ("messageRef", "operationRef"),
"signalEventDefinition": ("signalRef",),
"errorEventDefinition": ("errorRef",),
"escalationEventDefinition": ("escalationRef",),
"compensateEventDefinition": ("activityRef",),
}
fields = fields_by_element.get(element_type, ())
for field in fields:
value = attributes.get(field)
if value:
references.append((value, element_id, field))
def _qualified_name(tag: str) -> tuple[str | None, str]:
if tag.startswith("{") and "}" in tag:
namespace, local_name = tag[1:].split("}", 1)
return namespace, local_name
return None, tag
def _bounded_attribute(value: str | None, limit: int) -> str | None:
if value is None:
return None
text = value.strip()
return text[:limit] if text else None
__all__ = [
"BPMN_MODEL_NAMESPACE",
"BpmnInspectionError",
"NATIVE_EXECUTION_ELEMENTS",
"NATIVE_MAPPING_ELEMENTS",
"bpmn_support_level",
"inspect_bpmn_xml",
"parse_bpmn_xml",
]
from govoplan_workflow_engine.backend.bpmn import * # noqa: F401,F403
+2 -720
View File
@@ -1,721 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
import hashlib
import logging
from collections import Counter
from dataclasses import dataclass
from importlib.metadata import entry_points
from threading import Lock
from typing import Literal, Protocol, runtime_checkable
from xml.etree.ElementTree import Element
from govoplan_workflow.backend.bpmn import (
BPMN_DI_NAMESPACE,
BPMN_MODEL_NAMESPACE,
OMG_DC_NAMESPACE,
BpmnDiagnostic,
BpmnInspection,
inspect_bpmn_xml,
parse_bpmn_xml,
)
from govoplan_workflow.backend.bpmn_graph import (
BPMN_EDGE_LOCAL_NAMES,
BPMN_NODE_LOCAL_NAMES,
BpmnGraphError,
NATIVE_BPMN_ADAPTER_ID,
NATIVE_BPMN_ADAPTER_VERSION,
import_bpmn_graph,
runtime_diagnostics,
)
from govoplan_workflow.backend.schemas import (
WorkflowEdge,
WorkflowGraph,
WorkflowNode,
WorkflowPosition,
)
BPMN_ADAPTER_ENTRY_POINT_GROUP = "govoplan.workflow.bpmn_adapters"
INTERCHANGE_ADAPTER_ID = "bpmn.interchange"
NATIVE_LINEAR_ADAPTER_ID = "govoplan.native.linear"
RuntimeKind = Literal["model_only", "native_graph", "external"]
logger = logging.getLogger(__name__)
@dataclass(frozen=True, slots=True)
class BpmnExecutionProfile:
id: str
version: str
label: str
description: str
conformance: str
runtime_kind: RuntimeKind
executable: bool
supported_elements: tuple[str, ...]
supported_event_definitions: tuple[str, ...] = ()
requirements: tuple[str, ...] = ()
@runtime_checkable
class BpmnExecutionAdapter(Protocol):
profile: BpmnExecutionProfile
def diagnostics(
self,
xml: str,
inspection: BpmnInspection,
*,
activation: bool,
) -> tuple[BpmnDiagnostic, ...]:
...
def compile(
self,
xml: str,
inspection: BpmnInspection,
) -> WorkflowGraph | None:
...
class BpmnAdapterError(ValueError):
def __init__(self, diagnostics: tuple[BpmnDiagnostic, ...]) -> None:
self.diagnostics = diagnostics
first = next(
(item for item in diagnostics if item.severity == "error"),
diagnostics[0] if diagnostics else None,
)
super().__init__(
first.message if first is not None else "BPMN adapter validation failed."
)
class BpmnExecutionAdapterRegistry:
def __init__(self) -> None:
self._adapters: dict[tuple[str, str], BpmnExecutionAdapter] = {}
def register(self, adapter: BpmnExecutionAdapter) -> None:
if not isinstance(adapter, BpmnExecutionAdapter):
raise TypeError("BPMN adapter does not implement BpmnExecutionAdapter.")
profile = adapter.profile
if not profile.id.strip() or not profile.version.strip():
raise ValueError("BPMN adapters require a stable id and version.")
key = (profile.id, profile.version)
if key in self._adapters:
raise ValueError(
f"Duplicate BPMN execution adapter {profile.id}@{profile.version}."
)
self._adapters[key] = adapter
def profiles(self) -> tuple[BpmnExecutionProfile, ...]:
return tuple(
adapter.profile
for _key, adapter in sorted(
self._adapters.items(),
key=lambda item: (
not item[1].profile.executable,
item[1].profile.label.casefold(),
item[1].profile.id,
item[1].profile.version,
),
)
)
def resolve(
self,
adapter_id: str,
version: str | None = None,
) -> BpmnExecutionAdapter | None:
if version is not None:
return self._adapters.get((adapter_id, version))
matches = [
adapter
for (candidate_id, _candidate_version), adapter in self._adapters.items()
if candidate_id == adapter_id
]
return max(matches, key=lambda item: item.profile.version) if matches else None
def require(
self,
adapter_id: str,
version: str | None = None,
) -> BpmnExecutionAdapter:
adapter = self.resolve(adapter_id, version)
if adapter is None:
suffix = f"@{version}" if version else ""
raise BpmnAdapterError(
(
BpmnDiagnostic(
severity="error",
code="adapter.unavailable",
message=(
f"BPMN execution adapter {adapter_id}{suffix} is not "
"installed."
),
),
)
)
return adapter
class InterchangeOnlyAdapter:
profile = BpmnExecutionProfile(
id=INTERCHANGE_ADAPTER_ID,
version="1.0.0",
label="Historical model-only revision",
description=(
"Read historical BPMN 2.0 revisions that were stored without a "
"native graph or executable semantics."
),
conformance="BPMN 2.0 interchange",
runtime_kind="model_only",
executable=False,
supported_elements=("*",),
)
def diagnostics(
self,
xml: str,
inspection: BpmnInspection,
*,
activation: bool,
) -> tuple[BpmnDiagnostic, ...]:
del xml, inspection
if not activation:
return ()
return (
BpmnDiagnostic(
severity="error",
code="adapter.model_only",
message=(
"This BPMN revision is model-only. Select an executable "
"conformance profile and save a new revision before activation."
),
),
)
def compile(
self,
xml: str,
inspection: BpmnInspection,
) -> WorkflowGraph | None:
del xml, inspection
return None
class NativeBpmnGraphAdapter:
profile = BpmnExecutionProfile(
id=NATIVE_BPMN_ADAPTER_ID,
version=NATIVE_BPMN_ADAPTER_VERSION,
label="GovOPlaN native BPMN",
description=(
"Use BPMN 2.0 as the canonical GovOPlaN graph language. The "
"native graph preserves standard notation while activation "
"fails closed for runtime semantics that are not implemented."
),
conformance="BPMN 2.0 native graph profile 1",
runtime_kind="native_graph",
executable=True,
supported_elements=tuple(
sorted(
{
"definitions",
"process",
"collaboration",
"choreography",
"documentation",
"extensionElements",
"incoming",
"outgoing",
"conditionExpression",
"laneSet",
"flowNodeRef",
*BPMN_NODE_LOCAL_NAMES,
*BPMN_EDGE_LOCAL_NAMES,
}
)
),
supported_event_definitions=(
"message",
"timer",
"conditional",
"signal",
"error",
"escalation",
"compensation",
"link",
"cancel",
"terminate",
),
requirements=(
"BPMN is stored as the canonical native graph",
"Imported XML is parsed with bounded, entity-safe inspection",
"Unsupported execution semantics remain editable but block activation",
),
)
def diagnostics(
self,
xml: str,
inspection: BpmnInspection,
*,
activation: bool,
) -> tuple[BpmnDiagnostic, ...]:
diagnostics = [
item for item in inspection.diagnostics if item.severity == "error"
]
try:
graph = import_bpmn_graph(xml)
except BpmnGraphError as exc:
diagnostics.extend(exc.diagnostics)
return _deduplicate_diagnostics(diagnostics)
if activation:
diagnostics.extend(runtime_diagnostics(graph))
return _deduplicate_diagnostics(diagnostics)
def compile(
self,
xml: str,
inspection: BpmnInspection,
) -> WorkflowGraph | None:
del inspection
return import_bpmn_graph(xml)
class NativeLinearAdapter:
_supported_elements = frozenset(
{
"definitions",
"process",
"documentation",
"extensionElements",
"incoming",
"outgoing",
"startEvent",
"endEvent",
"task",
"manualTask",
"userTask",
"sequenceFlow",
}
)
_node_types = frozenset(
{"startEvent", "endEvent", "task", "manualTask", "userTask"}
)
profile = BpmnExecutionProfile(
id=NATIVE_LINEAR_ADAPTER_ID,
version="1.0.0",
label="GovOPlaN native linear",
description=(
"Execute a single linear process containing a plain start event, "
"human tasks, and one or more plain end events."
),
conformance="GovOPlaN native linear BPMN profile 1",
runtime_kind="native_graph",
executable=True,
supported_elements=tuple(sorted(_supported_elements)),
requirements=(
"Exactly one executable process",
"Exactly one plain start event",
"No gateways, subprocesses, boundary events, or event definitions",
"Every activity has one incoming and one outgoing sequence flow",
),
)
def diagnostics(
self,
xml: str,
inspection: BpmnInspection,
*,
activation: bool,
) -> tuple[BpmnDiagnostic, ...]:
del activation
diagnostics = list(
item
for item in inspection.diagnostics
if item.severity == "error"
)
unsupported = [
item
for item in inspection.elements
if item.element_type not in self._supported_elements
]
for item in unsupported:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="adapter.unsupported_element",
message=(
f"{item.element_type} is not supported by the "
f"{self.profile.label} profile."
),
element_id=item.element_id,
)
)
if inspection.process_count != 1:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="adapter.process_count",
message="The native linear profile requires exactly one process.",
)
)
if inspection.executable_process_count != 1:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="adapter.executable_process",
message=(
"The native linear profile requires one process with "
"isExecutable=\"true\"."
),
)
)
if diagnostics:
return _deduplicate_diagnostics(diagnostics)
root = parse_bpmn_xml(xml)
process = next(
(
item
for item in root
if _qualified_name(item.tag)
== (BPMN_MODEL_NAMESPACE, "process")
),
None,
)
if process is None:
return (
BpmnDiagnostic(
severity="error",
code="adapter.process_missing",
message="The executable BPMN process is missing.",
),
)
node_ids = {
item.attrib.get("id", "")
for item in process
if _qualified_name(item.tag)[1] in self._node_types
and item.attrib.get("id")
}
starts = [
item
for item in process
if _qualified_name(item.tag)[1] == "startEvent"
]
if len(starts) != 1:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="adapter.start_count",
message="The native linear profile requires exactly one start event.",
)
)
incoming: Counter[str] = Counter()
outgoing: Counter[str] = Counter()
for item in process:
if _qualified_name(item.tag)[1] != "sequenceFlow":
continue
source = item.attrib.get("sourceRef", "")
target = item.attrib.get("targetRef", "")
if source in node_ids:
outgoing[source] += 1
if target in node_ids:
incoming[target] += 1
for item in process:
_namespace, element_type = _qualified_name(item.tag)
if element_type not in self._node_types:
continue
element_id = item.attrib.get("id")
if not element_id:
continue
if element_type != "startEvent" and incoming[element_id] != 1:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="adapter.incoming_count",
message=(
f"{element_type} requires exactly one incoming "
"sequence flow in the native linear profile."
),
element_id=element_id,
)
)
if element_type != "endEvent" and outgoing[element_id] != 1:
diagnostics.append(
BpmnDiagnostic(
severity="error",
code="adapter.outgoing_count",
message=(
f"{element_type} requires exactly one outgoing "
"sequence flow in the native linear profile."
),
element_id=element_id,
)
)
return _deduplicate_diagnostics(diagnostics)
def compile(
self,
xml: str,
inspection: BpmnInspection,
) -> WorkflowGraph | None:
diagnostics = self.diagnostics(xml, inspection, activation=True)
if any(item.severity == "error" for item in diagnostics):
raise BpmnAdapterError(diagnostics)
root = parse_bpmn_xml(xml)
process = next(
item
for item in root
if _qualified_name(item.tag) == (BPMN_MODEL_NAMESPACE, "process")
)
positions = _diagram_positions(root)
raw_nodes = [
item
for item in process
if _qualified_name(item.tag)[1] in self._node_types
]
id_map = {
item.attrib["id"]: _graph_id(item.attrib["id"])
for item in raw_nodes
}
fallback_positions = {
item.attrib["id"]: WorkflowPosition(x=80 + index * 240, y=140)
for index, item in enumerate(_linear_node_order(process, raw_nodes))
}
nodes: list[WorkflowNode] = []
for item in raw_nodes:
element_type = _qualified_name(item.tag)[1]
bpmn_id = item.attrib["id"]
label = item.attrib.get("name", "").strip()
if element_type == "startEvent":
node_type = "workflow.start.manual"
config = {"input_schema_ref": ""}
label = label or "Start"
elif element_type == "endEvent":
node_type = "workflow.end.completed"
config = {"output_mapping": {}}
label = label or "Completed"
else:
node_type = "workflow.activity"
label = label or "Activity"
config = {
"title": label,
"instructions": "",
"assignee": "",
"due_after": "",
}
nodes.append(
WorkflowNode(
id=id_map[bpmn_id],
type=node_type,
label=label,
position=positions.get(bpmn_id, fallback_positions[bpmn_id]),
config=config,
)
)
edges = [
WorkflowEdge(
id=_graph_id(item.attrib["id"]),
source=id_map[item.attrib["sourceRef"]],
target=id_map[item.attrib["targetRef"]],
)
for item in process
if _qualified_name(item.tag)[1] == "sequenceFlow"
]
return WorkflowGraph(nodes=nodes, edges=edges)
def assess_bpmn_adapter(
xml: str,
*,
adapter_id: str,
adapter_version: str | None = None,
activation: bool = False,
registry: BpmnExecutionAdapterRegistry | None = None,
) -> tuple[
BpmnExecutionAdapter,
BpmnInspection,
tuple[BpmnDiagnostic, ...],
]:
active_registry = registry or bpmn_adapter_registry()
adapter = active_registry.require(adapter_id, adapter_version)
inspection = inspect_bpmn_xml(xml)
diagnostics = adapter.diagnostics(
xml,
inspection,
activation=activation,
)
return adapter, inspection, diagnostics
def compile_bpmn_to_graph(
xml: str,
*,
adapter_id: str,
adapter_version: str | None = None,
activation: bool = False,
registry: BpmnExecutionAdapterRegistry | None = None,
) -> tuple[BpmnExecutionAdapter, BpmnInspection, WorkflowGraph | None]:
adapter, inspection, diagnostics = assess_bpmn_adapter(
xml,
adapter_id=adapter_id,
adapter_version=adapter_version,
activation=activation,
registry=registry,
)
if any(item.severity == "error" for item in diagnostics):
raise BpmnAdapterError(diagnostics)
graph = adapter.compile(xml, inspection)
if adapter.profile.executable and graph is None:
raise BpmnAdapterError(
(
BpmnDiagnostic(
severity="error",
code="adapter.no_runtime_materialization",
message=(
f"{adapter.profile.label} did not produce executable "
"Workflow runtime materialization."
),
),
)
)
return adapter, inspection, graph
_registry: BpmnExecutionAdapterRegistry | None = None
_registry_lock = Lock()
def bpmn_adapter_registry() -> BpmnExecutionAdapterRegistry:
global _registry
if _registry is not None:
return _registry
with _registry_lock:
if _registry is not None:
return _registry
registry = BpmnExecutionAdapterRegistry()
registry.register(NativeBpmnGraphAdapter())
registry.register(NativeLinearAdapter())
registry.register(InterchangeOnlyAdapter())
for entry_point in entry_points(group=BPMN_ADAPTER_ENTRY_POINT_GROUP):
try:
loaded = entry_point.load()
candidate = (
loaded()
if callable(loaded) and not hasattr(loaded, "profile")
else loaded
)
registry.register(candidate)
except Exception:
logger.exception(
"Could not register BPMN execution adapter %s",
entry_point.name,
)
_registry = registry
return registry
def _diagram_positions(root: Element) -> dict[str, WorkflowPosition]:
positions: dict[str, WorkflowPosition] = {}
for item in root.iter():
if _qualified_name(item.tag) != (BPMN_DI_NAMESPACE, "BPMNShape"):
continue
element_id = item.attrib.get("bpmnElement")
if not element_id:
continue
bounds = next(
(
child
for child in item
if _qualified_name(child.tag) == (OMG_DC_NAMESPACE, "Bounds")
),
None,
)
if bounds is None:
continue
try:
positions[element_id] = WorkflowPosition(
x=float(bounds.attrib.get("x", "0")),
y=float(bounds.attrib.get("y", "0")),
)
except ValueError:
continue
return positions
def _linear_node_order(
process: Element,
nodes: list[Element],
) -> list[Element]:
by_id = {item.attrib["id"]: item for item in nodes}
target_by_source = {
item.attrib.get("sourceRef"): item.attrib.get("targetRef")
for item in process
if _qualified_name(item.tag)[1] == "sequenceFlow"
}
start = next(
(
item
for item in nodes
if _qualified_name(item.tag)[1] == "startEvent"
),
None,
)
ordered: list[Element] = []
seen: set[str] = set()
current = start
while current is not None:
current_id = current.attrib["id"]
if current_id in seen:
break
seen.add(current_id)
ordered.append(current)
current = by_id.get(target_by_source.get(current_id, ""))
ordered.extend(item for item in nodes if item.attrib["id"] not in seen)
return ordered
def _graph_id(value: str) -> str:
if len(value) <= 120:
return value
digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
return f"{value[:103]}-{digest}"
def _qualified_name(tag: str) -> tuple[str | None, str]:
if tag.startswith("{") and "}" in tag:
namespace, local_name = tag[1:].split("}", 1)
return namespace, local_name
return None, tag
def _deduplicate_diagnostics(
diagnostics: list[BpmnDiagnostic],
) -> tuple[BpmnDiagnostic, ...]:
seen: set[tuple[str, str | None, str]] = set()
result: list[BpmnDiagnostic] = []
for item in diagnostics:
key = (item.code, item.element_id, item.message)
if key in seen:
continue
seen.add(key)
result.append(item)
return tuple(result)
__all__ = [
"BPMN_ADAPTER_ENTRY_POINT_GROUP",
"INTERCHANGE_ADAPTER_ID",
"NATIVE_BPMN_ADAPTER_ID",
"NATIVE_LINEAR_ADAPTER_ID",
"BpmnAdapterError",
"BpmnExecutionAdapter",
"BpmnExecutionAdapterRegistry",
"BpmnExecutionProfile",
"RuntimeKind",
"assess_bpmn_adapter",
"bpmn_adapter_registry",
"compile_bpmn_to_graph",
]
from govoplan_workflow_engine.backend.bpmn_adapters import * # noqa: F401,F403
File diff suppressed because it is too large Load Diff
+2 -14
View File
@@ -1,15 +1,3 @@
from govoplan_workflow.backend.db.models import (
WorkflowDefinition,
WorkflowDefinitionRevision,
WorkflowInstance,
WorkflowInstanceEvent,
WorkflowInstanceStep,
)
"""Compatibility facade for the extracted Workflow Engine backend."""
__all__ = [
"WorkflowDefinition",
"WorkflowDefinitionRevision",
"WorkflowInstance",
"WorkflowInstanceEvent",
"WorkflowInstanceStep",
]
from govoplan_workflow_engine.backend.db import * # noqa: F401,F403
+2 -473
View File
@@ -1,474 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
import uuid
from datetime import datetime
from typing import Any
from sqlalchemy import (
Boolean,
DateTime,
ForeignKey,
Index,
Integer,
JSON,
String,
Text,
UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from govoplan_core.db.base import Base, TimestampMixin
def new_uuid() -> str:
return str(uuid.uuid4())
class WorkflowDefinition(Base, TimestampMixin):
__tablename__ = "workflow_definitions"
__table_args__ = (
UniqueConstraint(
"scope_key",
"definition_key",
name="uq_workflow_definition_key",
),
Index("ix_workflow_definitions_tenant_status", "tenant_id", "status"),
Index("ix_workflow_definitions_tenant_updated", "tenant_id", "updated_at"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
scope_type: Mapped[str] = mapped_column(
String(20),
default="tenant",
nullable=False,
index=True,
)
scope_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
scope_key: Mapped[str] = mapped_column(
String(80),
nullable=False,
index=True,
)
definition_kind: Mapped[str] = mapped_column(
String(20),
default="flow",
nullable=False,
index=True,
)
inherit_to_lower_scopes: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
allow_start: Mapped[bool] = mapped_column(
Boolean,
default=True,
nullable=False,
)
allow_reuse: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
allow_automation: Mapped[bool] = mapped_column(
Boolean,
default=False,
nullable=False,
)
derived_from_definition_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
derived_from_revision: Mapped[int | None] = mapped_column(
Integer,
nullable=True,
)
derived_from_hash: Mapped[str | None] = mapped_column(
String(64),
nullable=True,
)
derivation_provenance: Mapped[dict[str, Any]] = mapped_column(
JSON,
default=dict,
nullable=False,
)
definition_key: Mapped[str] = mapped_column(String(120), nullable=False)
name: Mapped[str] = mapped_column(String(300), nullable=False)
description: Mapped[str | None] = mapped_column(Text)
status: Mapped[str] = mapped_column(
String(32),
default="draft",
nullable=False,
index=True,
)
current_revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
active_revision: Mapped[int | None] = mapped_column(Integer, nullable=True)
metadata_: Mapped[dict[str, Any]] = mapped_column(
"metadata",
JSON,
default=dict,
nullable=False,
)
created_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
updated_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
deleted_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
nullable=True,
index=True,
)
revisions: Mapped[list["WorkflowDefinitionRevision"]] = relationship(
back_populates="definition",
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):
__tablename__ = "workflow_definition_revisions"
__table_args__ = (
UniqueConstraint(
"definition_id",
"revision",
name="uq_workflow_definition_revision",
),
Index(
"ix_workflow_definition_revisions_tenant_definition",
"tenant_id",
"definition_id",
),
Index(
"ix_workflow_definition_revisions_content_hash",
"tenant_id",
"content_hash",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
index=True,
)
definition_id: Mapped[str] = mapped_column(
ForeignKey("workflow_definitions.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
revision: Mapped[int] = mapped_column(Integer, nullable=False)
schema_version: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
graph: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
content_hash: Mapped[str] = mapped_column(String(64), nullable=False)
library_id: Mapped[str] = mapped_column(String(100), nullable=False)
library_version: Mapped[str] = mapped_column(String(40), nullable=False)
execution_mode: Mapped[str] = mapped_column(
String(20),
default="hybrid",
nullable=False,
)
view_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
view_revision_id: Mapped[str | None] = mapped_column(
String(36),
nullable=True,
)
bpmn_xml: Mapped[str | None] = mapped_column(Text, nullable=True)
bpmn_hash: Mapped[str | None] = mapped_column(
String(64),
nullable=True,
index=True,
)
bpmn_adapter_id: Mapped[str | None] = mapped_column(
String(120),
nullable=True,
)
bpmn_adapter_version: Mapped[str | None] = mapped_column(
String(40),
nullable=True,
)
bpmn_runtime_kind: Mapped[str | None] = mapped_column(
String(20),
nullable=True,
)
bpmn_executable: Mapped[bool | None] = mapped_column(
Boolean,
nullable=True,
)
created_by: Mapped[str | None] = mapped_column(
String(255),
nullable=True,
index=True,
)
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,
)
start_origin: Mapped[str] = mapped_column(
String(30),
default="user",
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",
]
from govoplan_workflow_engine.backend.db.models import * # noqa: F401,F403
+2 -322
View File
@@ -1,323 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
from collections.abc import Mapping
from typing import Literal
from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.policy import (
DefinitionGovernanceAction,
DefinitionGovernanceRequest,
DefinitionScopeRef,
PolicyDecision,
PolicySourceStep,
definition_governance_policy,
)
from govoplan_core.core.workflows import workflow_runtime_worker
from govoplan_workflow.backend.db.models import WorkflowDefinition
WorkflowAction = Literal[
"view",
"edit",
"start",
"reuse",
"derive",
"automate",
]
WORKFLOW_ACTIONS: tuple[WorkflowAction, ...] = (
"view",
"edit",
"start",
"reuse",
"derive",
"automate",
)
def normalize_definition_scope(
principal: ApiPrincipal,
*,
scope_type: str,
scope_id: str | None,
administrative: bool,
) -> tuple[str | None, str, str | None, str]:
clean_type = scope_type.strip().casefold()
clean_id = str(scope_id or "").strip() or None
if clean_type == "system":
if not has_scope(principal, "system:governance:write"):
raise PermissionError(
"System definitions require system governance permission."
)
if clean_id is not None:
raise ValueError("System definitions do not carry a scope ID.")
return None, "system", None, "system"
if clean_type == "tenant":
if clean_id not in {None, principal.tenant_id}:
raise PermissionError(
"Definitions can only be created for the active tenant."
)
return (
principal.tenant_id,
"tenant",
principal.tenant_id,
f"tenant:{principal.tenant_id}",
)
if clean_type == "group":
if not clean_id:
raise ValueError("Group definitions require a group ID.")
if clean_id not in principal.group_ids and not administrative:
raise PermissionError(
"Definitions can only be created for one of the actor's groups."
)
return principal.tenant_id, "group", clean_id, f"group:{clean_id}"
if clean_type == "user":
clean_id = clean_id or principal.account_id
if (
clean_id not in {principal.membership_id, principal.account_id}
and not administrative
):
raise PermissionError(
"Definitions can only be created for the current user."
)
if clean_id == principal.membership_id:
clean_id = principal.account_id
return principal.tenant_id, "user", clean_id, f"user:{clean_id}"
raise ValueError(
"Definition scope must be system, tenant, group, or user."
)
def definition_decision(
definition: WorkflowDefinition,
*,
principal: ApiPrincipal,
registry: object | None,
action: WorkflowAction,
) -> PolicyDecision:
policy_action: DefinitionGovernanceAction = (
"run" if action == "start" else action
)
request = DefinitionGovernanceRequest(
module_id="workflow",
definition_ref=f"workflow-definition:{definition.id}",
tenant_id=principal.tenant_id,
definition_scope=DefinitionScopeRef(
scope_type=definition.scope_type, # type: ignore[arg-type]
scope_id=definition.scope_id,
),
target_scope=_target_scope(definition, principal),
definition_kind=definition.definition_kind, # type: ignore[arg-type]
action=policy_action,
actor=principal.to_platform_principal(),
status=definition.status,
inherit_to_lower_scopes=definition.inherit_to_lower_scopes,
allow_run=definition.allow_start,
allow_reuse=definition.allow_reuse,
allow_automation=definition.allow_automation,
context=_ancestor_context(definition),
)
provider = definition_governance_policy(registry)
if provider is not None:
return provider.resolve_definition_action(request=request)
return _tenant_local_fallback(request, displayed_action=action)
def definition_governance_payload(
definition: WorkflowDefinition,
*,
principal: ApiPrincipal,
registry: object | None,
) -> dict[str, object]:
runtime_available = workflow_runtime_worker(registry) is not None
actions = {
action: definition_decision(
definition,
principal=principal,
registry=registry,
action=action,
).to_dict()
for action in WORKFLOW_ACTIONS
}
return {
"scope_type": definition.scope_type,
"scope_id": definition.scope_id,
"definition_kind": definition.definition_kind,
"inherit_to_lower_scopes": definition.inherit_to_lower_scopes,
"allow_start": definition.allow_start,
"allow_reuse": definition.allow_reuse,
"allow_automation": definition.allow_automation,
"derived_from_definition_id": (
definition.derived_from_definition_id
),
"derived_from_revision": definition.derived_from_revision,
"derived_from_hash": definition.derived_from_hash,
"derivation_provenance": dict(definition.derivation_provenance),
"actions": actions,
"automation_runtime_available": runtime_available,
"automation_runtime_reason": (
None
if runtime_available
else "Automatic reconciliation requires the Workflow runtime worker."
),
}
def require_definition_action(
definition: WorkflowDefinition,
*,
principal: ApiPrincipal,
registry: object | None,
action: WorkflowAction,
) -> PolicyDecision:
decision = definition_decision(
definition,
principal=principal,
registry=registry,
action=action,
)
if not decision.allowed:
raise PermissionError(
decision.reason or f"Definition action is not allowed: {action}"
)
return decision
def _target_scope(
definition: WorkflowDefinition,
principal: ApiPrincipal,
) -> DefinitionScopeRef:
if (
definition.scope_type == "group"
and definition.scope_id in principal.group_ids
):
return DefinitionScopeRef("group", definition.scope_id)
if definition.scope_type == "user" and definition.scope_id in {
principal.membership_id,
principal.account_id,
}:
return DefinitionScopeRef("user", definition.scope_id)
return DefinitionScopeRef("tenant", principal.tenant_id)
def _ancestor_context(
definition: WorkflowDefinition,
) -> dict[str, object]:
provenance = definition.derivation_provenance
limits = provenance.get("source_effective_limits")
source = provenance.get("source_scope")
context: dict[str, object] = {}
if isinstance(limits, Mapping):
context["ancestor_limits"] = {
"inherit_to_lower_scopes": bool(
limits.get("inherit_to_lower_scopes", True)
),
"allow_run": bool(limits.get("allow_start")),
"allow_reuse": bool(limits.get("allow_reuse")),
"allow_automation": bool(limits.get("allow_automation")),
}
if isinstance(source, Mapping):
context["ancestor_source"] = dict(source)
return context
def _tenant_local_fallback(
request: DefinitionGovernanceRequest,
*,
displayed_action: WorkflowAction,
) -> PolicyDecision:
ancestor = request.context.get("ancestor_limits")
ancestor_limits = ancestor if isinstance(ancestor, Mapping) else {}
effective_limits = {
"inherit_to_lower_scopes": (
request.inherit_to_lower_scopes
and _fallback_ancestor_flag(
ancestor_limits,
"inherit_to_lower_scopes",
)
),
"allow_run": request.allow_run
and _fallback_ancestor_flag(ancestor_limits, "allow_run"),
"allow_reuse": request.allow_reuse
and _fallback_ancestor_flag(ancestor_limits, "allow_reuse"),
"allow_automation": request.allow_automation
and _fallback_ancestor_flag(
ancestor_limits,
"allow_automation",
),
}
local = (
request.definition_scope.scope_type == "tenant"
and request.definition_scope.scope_id == request.tenant_id
and request.actor.tenant_id == request.tenant_id
)
allowed = False
reason: str | None = None
if not local:
reason = (
"Inherited definitions require the Policy module; only local "
"tenant definitions are available."
)
elif request.action in {"view", "edit"}:
allowed = True
elif request.action == "run":
allowed = (
request.definition_kind == "flow"
and request.status == "active"
and effective_limits["allow_run"]
)
reason = (
None
if allowed
else "Only active local flows with starting enabled can start."
)
else:
reason = "Definition reuse and automation require the Policy module."
return PolicyDecision(
allowed=allowed,
reason=reason,
source_path=(
PolicySourceStep(
scope_type=request.definition_scope.scope_type,
scope_id=request.definition_scope.scope_id,
label="Tenant-local conservative fallback",
applied_fields=(
"definition_kind",
"status",
"allow_run",
),
policy={
"policy_module_available": False,
"definition_kind": request.definition_kind,
"status": request.status,
"allow_start": request.allow_run,
},
),
),
requirements=(
()
if allowed
else (f"workflow.definition.{displayed_action}",)
),
details={
"fallback": "tenant_local",
"action": displayed_action,
"effective_limits": effective_limits,
},
)
def _fallback_ancestor_flag(
limits: Mapping[str, object],
key: str,
) -> bool:
value = limits.get(key)
return True if value is None else value is True
__all__ = [
"WORKFLOW_ACTIONS",
"definition_decision",
"definition_governance_payload",
"normalize_definition_scope",
"require_definition_action",
]
from govoplan_workflow_engine.backend.governance import * # noqa: F401,F403
File diff suppressed because it is too large Load Diff
+25 -248
View File
@@ -1,221 +1,55 @@
from __future__ import annotations
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,
)
from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE
from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
DocumentationTopic,
FrontendModule,
FrontendRoute,
MigrationSpec,
ModuleContext,
ModuleInterfaceProvider,
ModuleInterfaceRequirement,
ModuleManifest,
NavItem,
PermissionDefinition,
RoleTemplate,
ViewSurface,
)
from govoplan_core.core.policy import (
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
from govoplan_workflow_engine.backend.manifest import (
ADMIN_SCOPE,
DEFINITION_READ_SCOPE,
DEFINITION_WRITE_SCOPE,
INSTANCE_READ_SCOPE,
INSTANCE_START_SCOPE,
INSTANCE_TRANSITION_SCOPE,
)
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
MODULE_ID = "workflow"
MODULE_NAME = "Workflow"
MODULE_VERSION = "0.1.14"
DEFINITION_READ_SCOPE = "workflow:definition:read"
DEFINITION_WRITE_SCOPE = "workflow:definition:write"
INSTANCE_READ_SCOPE = "workflow:instance:read"
INSTANCE_START_SCOPE = "workflow:instance:start"
INSTANCE_TRANSITION_SCOPE = "workflow:instance:transition"
ADMIN_SCOPE = "workflow:instance:admin"
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
label=label,
description=description,
category="Workflow",
level="tenant",
module_id=module_id,
resource=resource,
action=action,
)
PERMISSIONS = (
_permission(
DEFINITION_READ_SCOPE,
"View workflow definitions",
"Read workflow graphs, versions, diagnostics, and referenced contracts.",
),
_permission(
DEFINITION_WRITE_SCOPE,
"Manage workflow definitions",
"Create, edit, validate, and publish workflow definitions.",
),
_permission(
INSTANCE_READ_SCOPE,
"View workflow instances",
"Read workflow progress, pending actions, and transition evidence.",
),
_permission(
INSTANCE_START_SCOPE,
"Start workflows",
"Start approved workflow definitions for authorized subjects.",
),
_permission(
INSTANCE_TRANSITION_SCOPE,
"Advance workflows",
"Complete activities and invoke authorized workflow transitions.",
),
_permission(
ADMIN_SCOPE,
"Administer workflows",
"Manage workflow definitions and instances across the tenant.",
),
)
ROLE_TEMPLATES = (
RoleTemplate(
slug="workflow_designer",
name="Workflow designer",
description="Design, validate, and publish workflow definitions.",
permissions=(DEFINITION_READ_SCOPE, DEFINITION_WRITE_SCOPE, INSTANCE_READ_SCOPE),
),
RoleTemplate(
slug="workflow_operator",
name="Workflow operator",
description="Start and advance approved workflows.",
permissions=(
DEFINITION_READ_SCOPE,
INSTANCE_READ_SCOPE,
INSTANCE_START_SCOPE,
INSTANCE_TRANSITION_SCOPE,
),
),
)
def _router(context: ModuleContext):
from govoplan_workflow.backend.runtime import configure_runtime
configure_runtime(registry=context.registry, settings=context.settings)
from govoplan_workflow.backend.router import router
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,
version=MODULE_VERSION,
dependencies=(),
optional_dependencies=(
"access",
"audit",
"dataflow",
"datasources",
"notifications",
"policy",
"tasks",
"views",
),
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,
),
dependencies=("workflow_engine",),
provides_interfaces=(
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),
ModuleInterfaceProvider(name="workflow.bpmn_interchange", version="1.0.0"),
ModuleInterfaceProvider(
name="workflow.bpmn_execution_adapters",
version="1.0.0",
),
ModuleInterfaceProvider(name="workflow.editor", version=MODULE_VERSION),
),
requires_interfaces=(
ModuleInterfaceRequirement(
name=CAPABILITY_ACCESS_REFERENCE_OPTIONS,
version_min="0.1.0",
version_max_exclusive="0.2.0",
optional=True,
),
ModuleInterfaceRequirement(
name="dataflow.run_lifecycle",
version_min="0.1.14",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="auth.automation_principal",
name="workflow.definition_graph",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name=CAPABILITY_NOTIFICATIONS_DISPATCH,
name="workflow.definition_catalogue",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="policy.definition_governance",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
),
ModuleInterfaceRequirement(
name="views.resolver",
version_min="0.1.0",
version_max_exclusive="1.0.0",
optional=True,
name="workflow.bpmn_interchange",
version_min="1.0.0",
version_max_exclusive="2.0.0",
),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
nav_items=(
NavItem(
path="/workflow",
@@ -255,83 +89,26 @@ 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",
),
retirement_notes=(
"Destructive retirement drops Workflow definitions and immutable revisions "
"after the installer captures a database snapshot."
),
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
workflow_models.WorkflowDefinition,
workflow_models.WorkflowDefinitionRevision,
workflow_models.WorkflowInstance,
workflow_models.WorkflowInstanceStep,
workflow_models.WorkflowInstanceEvent,
label="Workflow",
),
),
documentation=(
DocumentationTopic(
id="workflow.definition-graphs",
title="Workflow definition graphs",
summary="Governed process graphs built on the shared definition editor contract.",
body=(
"Workflow provides its own trigger, activity, decision, wait, integration, "
"and outcome node library on top of Core's domain-neutral graph contract. "
"Unlike Dataflow, Workflow permits cycles for correction and retry paths. "
"Module actions are addressed through versioned capabilities rather than "
"implementation imports. Definitions are persisted as immutable graph "
"revisions; activation pins the exact revision used by future instances."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "product_owner"),
related_modules=("dataflow", "datasources", "tasks", "notifications", "audit"),
order=76,
),
DocumentationTopic(
id="workflow.bpmn-interchange",
title="BPMN modeling and execution profiles",
id="workflow.editor",
title="Workflow editor and inspection workspace",
summary=(
"Lossless BPMN 2.0 revisions with explicit, fail-closed "
"execution conformance."
"Optional authoring, validation, revision inspection, and "
"operator controls for Workflow Engine."
),
body=(
"BPMN 2.0 is Workflow's canonical native graph language. The "
"shared graph editor models BPMN nodes, flows, containment, and "
"diagram geometry directly without a separate browser-side "
"modeler. Imported XML is normalized into the graph and every "
"immutable revision pins a deterministic XML artifact. Activation "
"requires a pinned execution adapter and version whose declared "
"profile accepts every modeled runtime semantic. Unsupported "
"runtime constructs remain editable and exportable. "
"Adapter packages integrate through the "
"govoplan.workflow.bpmn_adapters entry-point group and must "
"materialize canonical Workflow runtime state; Workflow never "
"imports a concrete engine module."
"Workflow adds the visual BPMN/native graph editor, definition "
"catalogue, immutable revision comparison, activation controls, "
"instance inspection, and governed override/reset experience. "
"Definitions and instances remain owned and executed by the "
"headless Workflow Engine module."
),
layer="available",
documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "product_owner"),
related_modules=("audit", "policy", "views"),
order=77,
related_modules=("workflow_engine", "views", "policy", "audit"),
order=76,
),
),
)
@@ -1 +0,0 @@
"""Workflow database migrations."""
@@ -1 +0,0 @@
"""Workflow Alembic revisions."""
@@ -1,201 +0,0 @@
"""v0.1.14 Workflow definitions
Revision ID: a7c4e2f9b1d3
Revises: None
Create Date: 2026-07-28 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "a7c4e2f9b1d3"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"workflow_definitions",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("definition_key", sa.String(length=120), nullable=False),
sa.Column("name", sa.String(length=300), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("status", sa.String(length=32), nullable=False),
sa.Column("current_revision", sa.Integer(), nullable=False),
sa.Column("active_revision", sa.Integer(), nullable=True),
sa.Column("metadata", sa.JSON(), nullable=False),
sa.Column("created_by", sa.String(length=255), nullable=True),
sa.Column("updated_by", sa.String(length=255), nullable=True),
sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("id", name=op.f("pk_workflow_definitions")),
sa.UniqueConstraint(
"tenant_id",
"definition_key",
name="uq_workflow_definition_key",
),
)
op.create_index(
op.f("ix_workflow_definitions_created_by"),
"workflow_definitions",
["created_by"],
unique=False,
)
op.create_index(
op.f("ix_workflow_definitions_deleted_at"),
"workflow_definitions",
["deleted_at"],
unique=False,
)
op.create_index(
op.f("ix_workflow_definitions_status"),
"workflow_definitions",
["status"],
unique=False,
)
op.create_index(
op.f("ix_workflow_definitions_tenant_id"),
"workflow_definitions",
["tenant_id"],
unique=False,
)
op.create_index(
op.f("ix_workflow_definitions_updated_by"),
"workflow_definitions",
["updated_by"],
unique=False,
)
op.create_index(
"ix_workflow_definitions_tenant_status",
"workflow_definitions",
["tenant_id", "status"],
unique=False,
)
op.create_index(
"ix_workflow_definitions_tenant_updated",
"workflow_definitions",
["tenant_id", "updated_at"],
unique=False,
)
op.create_table(
"workflow_definition_revisions",
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("revision", sa.Integer(), nullable=False),
sa.Column("schema_version", sa.Integer(), nullable=False),
sa.Column("graph", sa.JSON(), nullable=False),
sa.Column("content_hash", sa.String(length=64), nullable=False),
sa.Column("library_id", sa.String(length=100), nullable=False),
sa.Column("library_version", sa.String(length=40), nullable=False),
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"],
name=op.f(
"fk_workflow_definition_revisions_definition_id_workflow_definitions"
),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_workflow_definition_revisions"),
),
sa.UniqueConstraint(
"definition_id",
"revision",
name="uq_workflow_definition_revision",
),
)
op.create_index(
op.f("ix_workflow_definition_revisions_created_by"),
"workflow_definition_revisions",
["created_by"],
unique=False,
)
op.create_index(
op.f("ix_workflow_definition_revisions_definition_id"),
"workflow_definition_revisions",
["definition_id"],
unique=False,
)
op.create_index(
op.f("ix_workflow_definition_revisions_tenant_id"),
"workflow_definition_revisions",
["tenant_id"],
unique=False,
)
op.create_index(
"ix_workflow_definition_revisions_content_hash",
"workflow_definition_revisions",
["tenant_id", "content_hash"],
unique=False,
)
op.create_index(
"ix_workflow_definition_revisions_tenant_definition",
"workflow_definition_revisions",
["tenant_id", "definition_id"],
unique=False,
)
def downgrade() -> None:
op.drop_index(
"ix_workflow_definition_revisions_tenant_definition",
table_name="workflow_definition_revisions",
)
op.drop_index(
"ix_workflow_definition_revisions_content_hash",
table_name="workflow_definition_revisions",
)
op.drop_index(
op.f("ix_workflow_definition_revisions_tenant_id"),
table_name="workflow_definition_revisions",
)
op.drop_index(
op.f("ix_workflow_definition_revisions_definition_id"),
table_name="workflow_definition_revisions",
)
op.drop_index(
op.f("ix_workflow_definition_revisions_created_by"),
table_name="workflow_definition_revisions",
)
op.drop_table("workflow_definition_revisions")
op.drop_index(
"ix_workflow_definitions_tenant_updated",
table_name="workflow_definitions",
)
op.drop_index(
"ix_workflow_definitions_tenant_status",
table_name="workflow_definitions",
)
op.drop_index(
op.f("ix_workflow_definitions_updated_by"),
table_name="workflow_definitions",
)
op.drop_index(
op.f("ix_workflow_definitions_tenant_id"),
table_name="workflow_definitions",
)
op.drop_index(
op.f("ix_workflow_definitions_status"),
table_name="workflow_definitions",
)
op.drop_index(
op.f("ix_workflow_definitions_deleted_at"),
table_name="workflow_definitions",
)
op.drop_index(
op.f("ix_workflow_definitions_created_by"),
table_name="workflow_definitions",
)
op.drop_table("workflow_definitions")
@@ -1,209 +0,0 @@
"""v0.1.14 governed Workflow definitions
Revision ID: c6d8f1a3e5b7
Revises: a7c4e2f9b1d3
Create Date: 2026-07-28 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "c6d8f1a3e5b7"
down_revision = "a7c4e2f9b1d3"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("workflow_definitions") as batch_op:
batch_op.drop_constraint(
"uq_workflow_definition_key",
type_="unique",
)
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=True,
)
batch_op.add_column(
sa.Column(
"scope_type",
sa.String(length=20),
nullable=False,
server_default="tenant",
)
)
batch_op.add_column(
sa.Column("scope_id", sa.String(length=36), nullable=True)
)
batch_op.add_column(
sa.Column(
"scope_key",
sa.String(length=80),
nullable=False,
server_default="tenant:legacy",
)
)
batch_op.add_column(
sa.Column(
"definition_kind",
sa.String(length=20),
nullable=False,
server_default="flow",
)
)
batch_op.add_column(
sa.Column(
"inherit_to_lower_scopes",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.add_column(
sa.Column(
"allow_start",
sa.Boolean(),
nullable=False,
server_default=sa.true(),
)
)
batch_op.add_column(
sa.Column(
"allow_reuse",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.add_column(
sa.Column(
"allow_automation",
sa.Boolean(),
nullable=False,
server_default=sa.false(),
)
)
batch_op.add_column(
sa.Column(
"derived_from_definition_id",
sa.String(length=36),
nullable=True,
)
)
batch_op.add_column(
sa.Column("derived_from_revision", sa.Integer(), nullable=True)
)
batch_op.add_column(
sa.Column(
"derived_from_hash",
sa.String(length=64),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"derivation_provenance",
sa.JSON(),
nullable=False,
server_default=sa.text("'{}'"),
)
)
op.execute(
sa.text(
"UPDATE workflow_definitions "
"SET scope_id = tenant_id, "
"scope_key = 'tenant:' || tenant_id "
"WHERE tenant_id IS NOT NULL"
)
)
with op.batch_alter_table("workflow_definitions") as batch_op:
batch_op.create_unique_constraint(
"uq_workflow_definition_key",
["scope_key", "definition_key"],
)
for column in (
"scope_type",
"scope_id",
"scope_key",
"definition_kind",
"derived_from_definition_id",
):
op.create_index(
op.f(f"ix_workflow_definitions_{column}"),
"workflow_definitions",
[column],
unique=False,
)
with op.batch_alter_table(
"workflow_definition_revisions"
) as batch_op:
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=True,
)
def downgrade() -> None:
op.execute(
sa.text(
"UPDATE workflow_definition_revisions "
"SET tenant_id = COALESCE(tenant_id, 'system')"
)
)
with op.batch_alter_table(
"workflow_definition_revisions"
) as batch_op:
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=False,
)
for column in (
"derived_from_definition_id",
"definition_kind",
"scope_key",
"scope_id",
"scope_type",
):
op.drop_index(
op.f(f"ix_workflow_definitions_{column}"),
table_name="workflow_definitions",
)
op.execute(
sa.text(
"UPDATE workflow_definitions "
"SET tenant_id = COALESCE(tenant_id, 'system')"
)
)
with op.batch_alter_table("workflow_definitions") as batch_op:
batch_op.drop_constraint(
"uq_workflow_definition_key",
type_="unique",
)
batch_op.create_unique_constraint(
"uq_workflow_definition_key",
["tenant_id", "definition_key"],
)
batch_op.drop_column("derivation_provenance")
batch_op.drop_column("derived_from_hash")
batch_op.drop_column("derived_from_revision")
batch_op.drop_column("derived_from_definition_id")
batch_op.drop_column("allow_automation")
batch_op.drop_column("allow_reuse")
batch_op.drop_column("allow_start")
batch_op.drop_column("inherit_to_lower_scopes")
batch_op.drop_column("definition_kind")
batch_op.drop_column("scope_key")
batch_op.drop_column("scope_id")
batch_op.drop_column("scope_type")
batch_op.alter_column(
"tenant_id",
existing_type=sa.String(length=36),
nullable=False,
)
@@ -1,213 +0,0 @@
"""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")
@@ -1,68 +0,0 @@
"""v0.1.14 normalized BPMN revision artifacts
Revision ID: e9a4c6b8d2f1
Revises: d8f2a5c7e1b4
Create Date: 2026-07-30 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "e9a4c6b8d2f1"
down_revision = "d8f2a5c7e1b4"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("workflow_definition_revisions") as batch_op:
batch_op.add_column(sa.Column("bpmn_xml", sa.Text(), nullable=True))
batch_op.add_column(
sa.Column("bpmn_hash", sa.String(length=64), nullable=True)
)
batch_op.add_column(
sa.Column(
"bpmn_adapter_id",
sa.String(length=120),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"bpmn_adapter_version",
sa.String(length=40),
nullable=True,
)
)
batch_op.add_column(
sa.Column(
"bpmn_runtime_kind",
sa.String(length=20),
nullable=True,
)
)
batch_op.add_column(
sa.Column("bpmn_executable", sa.Boolean(), nullable=True)
)
op.create_index(
op.f("ix_workflow_definition_revisions_bpmn_hash"),
"workflow_definition_revisions",
["bpmn_hash"],
unique=False,
)
def downgrade() -> None:
op.drop_index(
op.f("ix_workflow_definition_revisions_bpmn_hash"),
table_name="workflow_definition_revisions",
)
with op.batch_alter_table("workflow_definition_revisions") as batch_op:
batch_op.drop_column("bpmn_executable")
batch_op.drop_column("bpmn_runtime_kind")
batch_op.drop_column("bpmn_adapter_version")
batch_op.drop_column("bpmn_adapter_id")
batch_op.drop_column("bpmn_hash")
batch_op.drop_column("bpmn_xml")
@@ -1,62 +0,0 @@
"""v0.1.14 workflow execution modes and focused Views
Revision ID: f1b7d3e5a9c2
Revises: e9a4c6b8d2f1
Create Date: 2026-07-30 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "f1b7d3e5a9c2"
down_revision = "e9a4c6b8d2f1"
branch_labels = None
depends_on = None
def upgrade() -> None:
with op.batch_alter_table("workflow_definition_revisions") as batch_op:
batch_op.add_column(
sa.Column(
"execution_mode",
sa.String(length=20),
nullable=False,
server_default="hybrid",
)
)
batch_op.add_column(
sa.Column("view_id", sa.String(length=36), nullable=True)
)
batch_op.add_column(
sa.Column(
"view_revision_id",
sa.String(length=36),
nullable=True,
)
)
with op.batch_alter_table("workflow_instances") as batch_op:
batch_op.add_column(
sa.Column(
"start_origin",
sa.String(length=30),
nullable=False,
server_default="user",
)
)
batch_op.create_index(
"ix_workflow_instances_start_origin",
["start_origin"],
unique=False,
)
def downgrade() -> None:
with op.batch_alter_table("workflow_instances") as batch_op:
batch_op.drop_index("ix_workflow_instances_start_origin")
batch_op.drop_column("start_origin")
with op.batch_alter_table("workflow_definition_revisions") as batch_op:
batch_op.drop_column("view_revision_id")
batch_op.drop_column("view_id")
batch_op.drop_column("execution_mode")
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+2 -10
View File
@@ -1,11 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
from govoplan_core.core.runtime import ModuleRuntimeState
_runtime = ModuleRuntimeState("Workflow")
configure_runtime = _runtime.configure_runtime
get_registry = _runtime.get_registry
get_settings = _runtime.get_settings
settings = _runtime.settings
from govoplan_workflow_engine.backend.runtime import * # noqa: F401,F403
+2 -551
View File
@@ -1,552 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
import math
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, Field, field_validator, model_validator
WorkflowDefinitionStatus = Literal["draft", "active", "archived"]
DefinitionScopeType = Literal["system", "tenant", "group", "user"]
DefinitionKind = Literal["flow", "template"]
BpmnRuntimeKind = Literal["model_only", "native_graph", "external"]
WorkflowExecutionMode = Literal["guided", "automated", "hybrid"]
WorkflowStartOrigin = Literal[
"user",
"api",
"schedule",
"event",
"parent_workflow",
"dependency",
"retry",
"replay",
"backfill",
]
BpmnSupportLevel = Literal[
"interchange_only",
"native_mapping",
"native_execution",
]
class WorkflowPosition(BaseModel):
x: float = 0
y: float = 0
@field_validator("x", "y")
@classmethod
def finite_coordinate(cls, value: float) -> float:
if not math.isfinite(value):
raise ValueError("Graph coordinates must be finite.")
return value
class WorkflowSize(BaseModel):
width: float = Field(default=100, gt=0, le=10_000)
height: float = Field(default=80, gt=0, le=10_000)
class WorkflowNode(BaseModel):
id: str = Field(min_length=1, max_length=120)
type: str = Field(min_length=1, max_length=120)
label: str = Field(default="", max_length=300)
position: WorkflowPosition = Field(default_factory=WorkflowPosition)
size: WorkflowSize | None = None
parent_id: str | None = Field(default=None, max_length=120)
process_id: str | None = Field(default=None, max_length=120)
config: dict[str, Any] = Field(default_factory=dict)
class WorkflowWaypoint(BaseModel):
x: float
y: float
@field_validator("x", "y")
@classmethod
def finite_coordinate(cls, value: float) -> float:
if not math.isfinite(value):
raise ValueError("Edge coordinates must be finite.")
return value
class WorkflowEdge(BaseModel):
id: str = Field(min_length=1, max_length=120)
type: Literal[
"bpmn.sequenceFlow",
"bpmn.messageFlow",
"bpmn.association",
"bpmn.dataInputAssociation",
"bpmn.dataOutputAssociation",
"bpmn.conversationLink",
] = "bpmn.sequenceFlow"
label: str = Field(default="", max_length=300)
source: str = Field(min_length=1, max_length=120)
target: str = Field(min_length=1, max_length=120)
source_port: str = Field(default="output", min_length=1, max_length=120)
target_port: str = Field(default="input", min_length=1, max_length=120)
config: dict[str, Any] = Field(default_factory=dict)
waypoints: list[WorkflowWaypoint] = Field(default_factory=list, max_length=500)
class WorkflowGraph(BaseModel):
schema_version: Literal[1] = 1
nodes: list[WorkflowNode] = Field(default_factory=list, max_length=150)
edges: list[WorkflowEdge] = Field(default_factory=list, max_length=300)
metadata: dict[str, Any] = Field(default_factory=dict)
class WorkflowGraphValidationRequest(BaseModel):
graph: WorkflowGraph
class WorkflowDiagnosticResponse(BaseModel):
severity: Literal["error", "warning"]
code: str
message: str
node_id: str | None = None
field: str | None = None
class WorkflowGraphValidationResponse(BaseModel):
valid: bool
diagnostics: list[WorkflowDiagnosticResponse]
class WorkflowPortResponse(BaseModel):
id: str
label: str
required: bool
multiple: bool
minimum_connections: int
class WorkflowConfigFieldResponse(BaseModel):
id: str
label: str
kind: str
required: bool
description: str | None
options: list[tuple[str, str]]
class WorkflowNodeTypeResponse(BaseModel):
type: str
category: str
category_label: str
label: str
description: str
icon: str
input_ports: list[WorkflowPortResponse]
output_ports: list[WorkflowPortResponse]
config_fields: list[WorkflowConfigFieldResponse]
default_config: dict[str, Any]
metadata: dict[str, Any] = Field(default_factory=dict)
class WorkflowNodeLibraryResponse(BaseModel):
id: str
version: str
allows_cycles: bool
nodes: list[WorkflowNodeTypeResponse]
class BpmnInspectionRequest(BaseModel):
xml: str = Field(min_length=1, max_length=1_048_576)
adapter_id: str = Field(
default="govoplan.native.bpmn",
min_length=1,
max_length=120,
)
adapter_version: str | None = Field(default=None, max_length=40)
activation: bool = False
class BpmnElementSupportResponse(BaseModel):
element_type: str
element_id: str | None = None
name: str | None = None
parent_type: str | None = None
parent_id: str | None = None
support_level: BpmnSupportLevel
class BpmnDiagnosticResponse(BaseModel):
severity: Literal["error", "warning", "info"]
code: str
message: str
element_id: str | None = None
class BpmnInspectionResponse(BaseModel):
valid_xml: bool
definitions_id: str | None = None
target_namespace: str | None = None
process_count: int
executable_process_count: int
collaboration_count: int
choreography_count: int
element_counts: dict[str, int]
support_counts: dict[str, int]
elements: list[BpmnElementSupportResponse]
diagnostics: list[BpmnDiagnosticResponse]
adapter_id: str | None = None
adapter_version: str | None = None
runtime_kind: BpmnRuntimeKind | None = None
executable: bool = False
activatable: bool = False
class BpmnAdapterProfileResponse(BaseModel):
id: str
version: str
label: str
description: str
conformance: str
runtime_kind: BpmnRuntimeKind
executable: bool
supported_elements: list[str]
supported_event_definitions: list[str]
requirements: list[str]
class BpmnSupportProfileResponse(BaseModel):
specification: str
model_namespace: str
interchange: str
native_runtime: str
native_execution_elements: list[str]
native_mapping_elements: list[str]
adapters: list[BpmnAdapterProfileResponse] = Field(default_factory=list)
class BpmnRevisionInput(BaseModel):
xml: str = Field(min_length=1, max_length=1_048_576)
adapter_id: str = Field(
default="govoplan.native.bpmn",
min_length=1,
max_length=120,
)
adapter_version: str | None = Field(default=None, max_length=40)
class BpmnRevisionSummaryResponse(BaseModel):
format: Literal["bpmn-2.0"] = "bpmn-2.0"
content_hash: str
adapter_id: str
adapter_version: str
runtime_kind: BpmnRuntimeKind
executable: bool
adapter_available: bool
class BpmnRevisionDocumentResponse(BpmnRevisionSummaryResponse):
definition_id: str
revision: int
xml: str
inspection: BpmnInspectionResponse
class BpmnCompileRequest(BaseModel):
xml: str = Field(min_length=1, max_length=1_048_576)
adapter_id: str = Field(
default="govoplan.native.bpmn",
min_length=1,
max_length=120,
)
adapter_version: str | None = Field(default=None, max_length=40)
class BpmnCompileResponse(BaseModel):
adapter: BpmnAdapterProfileResponse
graph: WorkflowGraph
inspection: BpmnInspectionResponse
class BpmnRenderRequest(BaseModel):
graph: WorkflowGraph
name: str = Field(default="", max_length=300)
class BpmnRenderResponse(BaseModel):
xml: str
inspection: BpmnInspectionResponse
class WorkflowDefinitionRevisionResponse(BaseModel):
id: str
revision: int
schema_version: int
graph: WorkflowGraph
content_hash: str
library_id: str
library_version: str
execution_mode: WorkflowExecutionMode
view_id: str | None = None
view_revision_id: str | None = None
bpmn: BpmnRevisionSummaryResponse | None = None
created_by: str | None
created_at: datetime
class WorkflowActionDecisionResponse(BaseModel):
allowed: bool
reason: str | None = None
source_path: list[dict[str, Any]] = Field(default_factory=list)
requirements: list[str] = Field(default_factory=list)
details: dict[str, Any] = Field(default_factory=dict)
class WorkflowGovernanceResponse(BaseModel):
scope_type: DefinitionScopeType
scope_id: str | None
definition_kind: DefinitionKind
inherit_to_lower_scopes: bool
allow_start: bool
allow_reuse: bool
allow_automation: bool
derived_from_definition_id: str | None
derived_from_revision: int | None
derived_from_hash: str | None
derivation_provenance: dict[str, Any] = Field(default_factory=dict)
actions: dict[str, WorkflowActionDecisionResponse]
automation_runtime_available: bool = False
automation_runtime_reason: str | None = None
class WorkflowDefinitionResponse(BaseModel):
id: str
tenant_id: str | None
key: str
name: str
description: str | None
status: WorkflowDefinitionStatus
current_revision: int
active_revision: int | None
metadata: dict[str, Any]
created_by: str | None
updated_by: str | None
created_at: datetime
updated_at: datetime
revision: WorkflowDefinitionRevisionResponse
governance: WorkflowGovernanceResponse
class WorkflowDefinitionListResponse(BaseModel):
definitions: list[WorkflowDefinitionResponse]
class WorkflowDefinitionRevisionListResponse(BaseModel):
revisions: list[WorkflowDefinitionRevisionResponse]
class WorkflowDefinitionCreateRequest(BaseModel):
key: str | None = Field(
default=None,
min_length=1,
max_length=120,
pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]*$",
)
name: str = Field(min_length=1, max_length=300)
description: str | None = Field(default=None, max_length=4_000)
graph: WorkflowGraph
bpmn: BpmnRevisionInput | None = None
metadata: dict[str, Any] = Field(default_factory=dict)
scope_type: DefinitionScopeType = "tenant"
scope_id: str | None = Field(default=None, max_length=36)
definition_kind: DefinitionKind = "flow"
inherit_to_lower_scopes: bool = False
allow_start: bool = True
allow_reuse: bool = False
allow_automation: bool = False
execution_mode: WorkflowExecutionMode = "hybrid"
view_id: str | None = Field(default=None, min_length=1, max_length=36)
view_revision_id: str | None = Field(
default=None,
min_length=1,
max_length=36,
)
@model_validator(mode="after")
def validate_view_pin(self):
if self.view_revision_id and not self.view_id:
raise ValueError("A pinned View revision requires a View")
return self
class WorkflowDefinitionUpdateRequest(BaseModel):
name: str = Field(min_length=1, max_length=300)
description: str | None = Field(default=None, max_length=4_000)
graph: WorkflowGraph
bpmn: BpmnRevisionInput | None = None
metadata: dict[str, Any] = Field(default_factory=dict)
expected_revision: int = Field(ge=1)
scope_type: DefinitionScopeType = "tenant"
scope_id: str | None = Field(default=None, max_length=36)
definition_kind: DefinitionKind = "flow"
inherit_to_lower_scopes: bool = False
allow_start: bool = True
allow_reuse: bool = False
allow_automation: bool = False
execution_mode: WorkflowExecutionMode = "hybrid"
view_id: str | None = Field(default=None, min_length=1, max_length=36)
view_revision_id: str | None = Field(
default=None,
min_length=1,
max_length=36,
)
@model_validator(mode="after")
def validate_view_pin(self):
if self.view_revision_id and not self.view_id:
raise ValueError("A pinned View revision requires a View")
return self
class WorkflowDefinitionDeriveRequest(BaseModel):
key: str | None = Field(
default=None,
min_length=1,
max_length=120,
pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]*$",
)
name: str = Field(min_length=1, max_length=300)
description: str | None = Field(default=None, max_length=4_000)
source_revision: int | None = Field(default=None, ge=1)
metadata: dict[str, Any] = Field(default_factory=dict)
scope_type: DefinitionScopeType = "tenant"
scope_id: str | None = Field(default=None, max_length=36)
definition_kind: DefinitionKind = "flow"
inherit_to_lower_scopes: bool = False
allow_start: bool = True
allow_reuse: bool = False
allow_automation: bool = False
execution_mode: WorkflowExecutionMode | None = None
view_id: str | None = Field(default=None, min_length=1, max_length=36)
view_revision_id: str | None = Field(
default=None,
min_length=1,
max_length=36,
)
@model_validator(mode="after")
def validate_view_pin(self):
if self.view_revision_id and not self.view_id:
raise ValueError("A pinned View revision requires a View")
return self
class WorkflowDefinitionActivateRequest(BaseModel):
revision: int | None = Field(default=None, ge=1)
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 WorkflowViewContextResponse(BaseModel):
view_id: str
revision_id: str | None = None
visible_surface_ids: list[str] = Field(default_factory=list)
step_id: str | None = None
node_id: str | None = None
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
execution_mode: WorkflowExecutionMode
start_origin: WorkflowStartOrigin
view_context: WorkflowViewContextResponse | None = None
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]
from govoplan_workflow_engine.backend.schemas import * # noqa: F401,F403
+2 -969
View File
@@ -1,970 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
from collections.abc import Mapping
from dataclasses import dataclass
import hashlib
import json
import re
import unicodedata
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.db.base import utcnow
from govoplan_workflow.backend.db.models import (
WorkflowDefinition,
WorkflowDefinitionRevision,
)
from govoplan_workflow.backend.node_library import WORKFLOW_GRAPH_LIBRARY
from govoplan_workflow.backend.bpmn import (
BpmnDiagnostic,
BpmnInspectionError,
)
from govoplan_workflow.backend.bpmn_adapters import (
RuntimeKind,
bpmn_adapter_registry,
)
from govoplan_workflow.backend.bpmn_graph import (
BpmnGraphError,
NATIVE_BPMN_ADAPTER_ID,
NATIVE_BPMN_ADAPTER_VERSION,
canonical_bpmn_graph,
export_bpmn_graph,
import_bpmn_graph,
materialize_runtime_graph,
runtime_diagnostics,
)
from govoplan_workflow.backend.governance import (
definition_governance_payload,
require_definition_action,
)
from govoplan_workflow.backend.schemas import (
BpmnRevisionInput,
BpmnRevisionSummaryResponse,
WorkflowDefinitionCreateRequest,
WorkflowDefinitionDeriveRequest,
WorkflowDefinitionResponse,
WorkflowDefinitionRevisionResponse,
WorkflowDefinitionUpdateRequest,
WorkflowGraph,
)
from govoplan_workflow.backend.validation import validate_workflow_graph
class WorkflowError(RuntimeError):
pass
class WorkflowNotFoundError(WorkflowError):
pass
class WorkflowConflictError(WorkflowError):
pass
class WorkflowValidationError(WorkflowError):
def __init__(self, diagnostics: tuple[object, ...]) -> None:
first = diagnostics[0] if diagnostics else None
super().__init__(
str(getattr(first, "message", "Workflow definition validation failed."))
)
self.diagnostics = diagnostics
class WorkflowBpmnValidationError(WorkflowError):
def __init__(self, diagnostics: tuple[BpmnDiagnostic, ...]) -> None:
first = next(
(item for item in diagnostics if item.severity == "error"),
diagnostics[0] if diagnostics else None,
)
super().__init__(
first.message if first is not None else "BPMN validation failed."
)
self.diagnostics = diagnostics
@dataclass(frozen=True, slots=True)
class _BpmnArtifact:
xml: str
content_hash: str
adapter_id: str
adapter_version: str
runtime_kind: RuntimeKind
executable: bool
def list_definitions(
session: Session,
*,
tenant_id: str,
) -> list[WorkflowDefinition]:
return list(
session.scalars(
select(WorkflowDefinition)
.where(
or_(
WorkflowDefinition.tenant_id == tenant_id,
WorkflowDefinition.tenant_id.is_(None),
),
WorkflowDefinition.deleted_at.is_(None),
)
.order_by(
WorkflowDefinition.updated_at.desc(),
WorkflowDefinition.name.asc(),
)
)
)
def get_definition(
session: Session,
*,
tenant_id: str,
definition_id: str,
) -> WorkflowDefinition:
definition = session.scalar(
select(WorkflowDefinition).where(
WorkflowDefinition.id == definition_id,
or_(
WorkflowDefinition.tenant_id == tenant_id,
WorkflowDefinition.tenant_id.is_(None),
),
WorkflowDefinition.deleted_at.is_(None),
)
)
if definition is None:
raise WorkflowNotFoundError("Workflow definition not found.")
return definition
def get_definition_revision(
session: Session,
*,
definition: WorkflowDefinition,
revision: int | None = None,
) -> WorkflowDefinitionRevision:
revision_number = revision or definition.current_revision
item = session.scalar(
select(WorkflowDefinitionRevision).where(
WorkflowDefinitionRevision.definition_id == definition.id,
WorkflowDefinitionRevision.tenant_id == definition.tenant_id,
WorkflowDefinitionRevision.revision == revision_number,
)
)
if item is None:
raise WorkflowNotFoundError("Workflow definition revision not found.")
return item
def list_definition_revisions(
session: Session,
*,
definition: WorkflowDefinition,
) -> list[WorkflowDefinitionRevision]:
return list(
session.scalars(
select(WorkflowDefinitionRevision)
.where(
WorkflowDefinitionRevision.definition_id == definition.id,
WorkflowDefinitionRevision.tenant_id == definition.tenant_id,
)
.order_by(WorkflowDefinitionRevision.revision.desc())
)
)
def create_definition(
session: Session,
*,
tenant_id: str,
actor_id: str | None,
payload: WorkflowDefinitionCreateRequest,
) -> WorkflowDefinition:
graph, bpmn = _prepare_revision_content(
graph=payload.graph,
bpmn=payload.bpmn,
)
stored_tenant_id = (
None if payload.scope_type == "system" else tenant_id
)
scope_id = (
None
if payload.scope_type == "system"
else tenant_id
if payload.scope_type == "tenant"
else payload.scope_id
)
scope_key = (
"system"
if payload.scope_type == "system"
else f"{payload.scope_type}:{scope_id}"
)
definition = WorkflowDefinition(
tenant_id=stored_tenant_id,
scope_type=payload.scope_type,
scope_id=scope_id,
scope_key=scope_key,
definition_kind=payload.definition_kind,
inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
allow_start=payload.allow_start,
allow_reuse=payload.allow_reuse,
allow_automation=payload.allow_automation,
definition_key=_available_key(
session,
scope_key=scope_key,
requested=payload.key,
name=payload.name,
),
name=payload.name.strip(),
description=_clean_optional(payload.description),
status="draft",
current_revision=1,
active_revision=None,
metadata_=dict(payload.metadata),
created_by=actor_id,
updated_by=actor_id,
)
definition.revisions.append(
_new_revision(
tenant_id=stored_tenant_id,
revision=1,
graph=graph,
bpmn=bpmn,
execution_mode=payload.execution_mode,
view_id=payload.view_id,
view_revision_id=payload.view_revision_id,
actor_id=actor_id,
)
)
session.add(definition)
session.flush()
return definition
def update_definition(
session: Session,
*,
tenant_id: str,
definition_id: str,
actor_id: str | None,
payload: WorkflowDefinitionUpdateRequest,
) -> WorkflowDefinition:
definition = get_definition(
session,
tenant_id=tenant_id,
definition_id=definition_id,
)
if payload.expected_revision != definition.current_revision:
raise WorkflowConflictError(
"Workflow definition changed on the server; "
f"expected revision {payload.expected_revision}, "
f"current revision is {definition.current_revision}."
)
if (
payload.scope_type != definition.scope_type
or payload.scope_id != definition.scope_id
and not (
definition.scope_type == "tenant"
and payload.scope_id in {None, definition.scope_id}
)
):
raise WorkflowConflictError(
"Definition scope is immutable; derive a scoped copy instead."
)
if payload.definition_kind != definition.definition_kind:
raise WorkflowConflictError(
"Definition kind is immutable; derive a flow or template instead."
)
current = get_definition_revision(session, definition=definition)
graph, bpmn = _prepare_revision_content(
graph=payload.graph,
bpmn=payload.bpmn,
)
definition.name = payload.name.strip()
definition.description = _clean_optional(payload.description)
definition.metadata_ = dict(payload.metadata)
ancestor_limits = _ancestor_governance_limits(
definition.derivation_provenance
)
definition.inherit_to_lower_scopes = (
payload.inherit_to_lower_scopes
and ancestor_limits["inherit_to_lower_scopes"]
)
definition.allow_start = (
payload.allow_start and ancestor_limits["allow_start"]
)
definition.allow_reuse = (
payload.allow_reuse and ancestor_limits["allow_reuse"]
)
definition.allow_automation = (
payload.allow_automation and ancestor_limits["allow_automation"]
)
definition.updated_by = actor_id
if not _revision_matches(
current,
graph=graph,
bpmn=bpmn,
execution_mode=payload.execution_mode,
view_id=payload.view_id,
view_revision_id=payload.view_revision_id,
):
definition.current_revision += 1
definition.revisions.append(
_new_revision(
tenant_id=definition.tenant_id,
revision=definition.current_revision,
graph=graph,
bpmn=bpmn,
execution_mode=payload.execution_mode,
view_id=payload.view_id,
view_revision_id=payload.view_revision_id,
actor_id=actor_id,
)
)
if definition.status == "active":
definition.status = "draft"
session.flush()
return definition
def derive_definition(
session: Session,
*,
tenant_id: str,
actor_id: str | None,
principal: ApiPrincipal,
registry: object | None,
source_definition_id: str,
payload: WorkflowDefinitionDeriveRequest,
) -> WorkflowDefinition:
source = get_definition(
session,
tenant_id=tenant_id,
definition_id=source_definition_id,
)
decision = require_definition_action(
source,
principal=principal,
registry=registry,
action="derive",
)
source_revision = get_definition_revision(
session,
definition=source,
revision=payload.source_revision,
)
stored_tenant_id = (
None if payload.scope_type == "system" else tenant_id
)
scope_id = (
None
if payload.scope_type == "system"
else tenant_id
if payload.scope_type == "tenant"
else payload.scope_id
)
scope_key = (
"system"
if payload.scope_type == "system"
else f"{payload.scope_type}:{scope_id}"
)
source_limits = _effective_governance_limits(
source,
decision_details=decision.details,
)
limits = {
"inherit_to_lower_scopes": (
source_limits["inherit_to_lower_scopes"]
and payload.inherit_to_lower_scopes
),
"allow_start": (
source_limits["allow_start"] and payload.allow_start
),
"allow_reuse": (
source_limits["allow_reuse"] and payload.allow_reuse
),
"allow_automation": (
source_limits["allow_automation"]
and payload.allow_automation
),
}
provenance = {
"source_ref": f"workflow-definition:{source.id}",
"source_scope": {
"scope_type": source.scope_type,
"scope_id": source.scope_id,
},
"source_definition_kind": source.definition_kind,
"source_revision": source_revision.revision,
"source_hash": source_revision.content_hash,
"source_effective_limits": limits,
"policy_decision": decision.to_dict(),
"derived_by": actor_id,
"derived_at": utcnow().isoformat(),
}
execution_mode = (
payload.execution_mode
if "execution_mode" in payload.model_fields_set
else source_revision.execution_mode
)
view_id = (
payload.view_id
if "view_id" in payload.model_fields_set
else source_revision.view_id
)
view_revision_id = (
payload.view_revision_id
if (
"view_revision_id" in payload.model_fields_set
or "view_id" in payload.model_fields_set
)
else source_revision.view_revision_id
)
if view_id is None:
view_revision_id = None
source_graph, source_bpmn = _prepare_revision_content(
graph=WorkflowGraph.model_validate(source_revision.graph),
bpmn=None,
)
definition = WorkflowDefinition(
tenant_id=stored_tenant_id,
scope_type=payload.scope_type,
scope_id=scope_id,
scope_key=scope_key,
definition_kind=payload.definition_kind,
inherit_to_lower_scopes=limits["inherit_to_lower_scopes"],
allow_start=limits["allow_start"],
allow_reuse=limits["allow_reuse"],
allow_automation=limits["allow_automation"],
derived_from_definition_id=source.id,
derived_from_revision=source_revision.revision,
derived_from_hash=source_revision.content_hash,
derivation_provenance=provenance,
definition_key=_available_key(
session,
scope_key=scope_key,
requested=payload.key,
name=payload.name,
),
name=payload.name.strip(),
description=_clean_optional(payload.description),
status="draft",
current_revision=1,
active_revision=None,
metadata_=dict(payload.metadata),
created_by=actor_id,
updated_by=actor_id,
)
definition.revisions.append(
WorkflowDefinitionRevision(
tenant_id=stored_tenant_id,
revision=1,
schema_version=source_revision.schema_version,
graph=_canonical_graph(source_graph),
content_hash=_content_hash(
source_graph,
bpmn=source_bpmn,
execution_mode=execution_mode,
view_id=view_id,
view_revision_id=view_revision_id,
),
library_id=source_revision.library_id,
library_version=source_revision.library_version,
execution_mode=execution_mode,
view_id=view_id,
view_revision_id=view_revision_id,
bpmn_xml=source_bpmn.xml,
bpmn_hash=source_bpmn.content_hash,
bpmn_adapter_id=source_bpmn.adapter_id,
bpmn_adapter_version=source_bpmn.adapter_version,
bpmn_runtime_kind=source_bpmn.runtime_kind,
bpmn_executable=source_bpmn.executable,
created_by=actor_id,
)
)
session.add(definition)
session.flush()
return definition
def activate_definition(
session: Session,
*,
tenant_id: str,
definition_id: str,
actor_id: str | None,
revision: int | None = None,
) -> WorkflowDefinition:
definition = get_definition(
session,
tenant_id=tenant_id,
definition_id=definition_id,
)
selected = get_definition_revision(
session,
definition=definition,
revision=revision,
)
if definition.definition_kind == "template":
raise WorkflowConflictError(
"Workflow templates cannot be activated or started."
)
_validated_graph(WorkflowGraph.model_validate(selected.graph))
_validate_bpmn_activation(selected)
_validate_execution_mode_activation(selected)
definition.active_revision = selected.revision
definition.status = "active"
definition.updated_by = actor_id
session.flush()
return definition
def archive_definition(
session: Session,
*,
tenant_id: str,
definition_id: str,
actor_id: str | None,
) -> WorkflowDefinition:
definition = get_definition(
session,
tenant_id=tenant_id,
definition_id=definition_id,
)
definition.status = "archived"
definition.updated_by = actor_id
session.flush()
return definition
def delete_definition(
session: Session,
*,
tenant_id: str,
definition_id: str,
actor_id: str | None,
) -> WorkflowDefinition:
definition = get_definition(
session,
tenant_id=tenant_id,
definition_id=definition_id,
)
definition.deleted_at = utcnow()
definition.updated_by = actor_id
session.flush()
return definition
def definition_response(
session: Session,
definition: WorkflowDefinition,
*,
principal: ApiPrincipal,
registry: object | None,
revision: int | None = None,
) -> WorkflowDefinitionResponse:
selected = get_definition_revision(
session,
definition=definition,
revision=revision,
)
return WorkflowDefinitionResponse(
id=definition.id,
tenant_id=definition.tenant_id,
key=definition.definition_key,
name=definition.name,
description=definition.description,
status=definition.status,
current_revision=definition.current_revision,
active_revision=definition.active_revision,
metadata=dict(definition.metadata_),
created_by=definition.created_by,
updated_by=definition.updated_by,
created_at=definition.created_at,
updated_at=definition.updated_at,
revision=revision_response(selected),
governance=definition_governance_payload(
definition,
principal=principal,
registry=registry,
),
)
def revision_response(
revision: WorkflowDefinitionRevision,
) -> WorkflowDefinitionRevisionResponse:
return WorkflowDefinitionRevisionResponse(
id=revision.id,
revision=revision.revision,
schema_version=revision.schema_version,
graph=canonical_bpmn_graph(
WorkflowGraph.model_validate(revision.graph)
),
content_hash=revision.content_hash,
library_id=revision.library_id,
library_version=revision.library_version,
execution_mode=revision.execution_mode,
view_id=revision.view_id,
view_revision_id=revision.view_revision_id,
bpmn=revision_bpmn_summary(revision),
created_by=revision.created_by,
created_at=revision.created_at,
)
def _new_revision(
*,
tenant_id: str | None,
revision: int,
graph: WorkflowGraph,
bpmn: _BpmnArtifact | None,
execution_mode: str,
view_id: str | None,
view_revision_id: str | None,
actor_id: str | None,
) -> WorkflowDefinitionRevision:
return WorkflowDefinitionRevision(
tenant_id=tenant_id,
revision=revision,
schema_version=graph.schema_version,
graph=_canonical_graph(graph),
content_hash=_content_hash(
graph,
bpmn=bpmn,
execution_mode=execution_mode,
view_id=view_id,
view_revision_id=view_revision_id,
),
library_id=WORKFLOW_GRAPH_LIBRARY.id,
library_version=WORKFLOW_GRAPH_LIBRARY.version,
execution_mode=execution_mode,
view_id=view_id,
view_revision_id=view_revision_id,
bpmn_xml=bpmn.xml if bpmn else None,
bpmn_hash=bpmn.content_hash if bpmn else None,
bpmn_adapter_id=bpmn.adapter_id if bpmn else None,
bpmn_adapter_version=bpmn.adapter_version if bpmn else None,
bpmn_runtime_kind=bpmn.runtime_kind if bpmn else None,
bpmn_executable=bpmn.executable if bpmn else None,
created_by=actor_id,
)
def revision_bpmn_summary(
revision: WorkflowDefinitionRevision,
) -> BpmnRevisionSummaryResponse | None:
if not revision.bpmn_xml:
return None
adapter_id = revision.bpmn_adapter_id or "bpmn.interchange"
adapter_version = revision.bpmn_adapter_version or "1.0.0"
adapter = bpmn_adapter_registry().resolve(adapter_id, adapter_version)
runtime_kind: RuntimeKind = "model_only"
if revision.bpmn_runtime_kind in {"model_only", "native_graph", "external"}:
runtime_kind = revision.bpmn_runtime_kind
return BpmnRevisionSummaryResponse(
content_hash=revision.bpmn_hash
or hashlib.sha256(revision.bpmn_xml.encode("utf-8")).hexdigest(),
adapter_id=adapter_id,
adapter_version=adapter_version,
runtime_kind=runtime_kind,
executable=bool(revision.bpmn_executable),
adapter_available=adapter is not None,
)
def _prepare_revision_content(
*,
graph: WorkflowGraph,
bpmn: BpmnRevisionInput | None,
) -> tuple[WorkflowGraph, _BpmnArtifact | None]:
try:
effective_graph = (
import_bpmn_graph(bpmn.xml)
if bpmn is not None
else canonical_bpmn_graph(graph)
)
except WorkflowBpmnValidationError:
raise
except (BpmnGraphError, BpmnInspectionError) as exc:
diagnostics = getattr(exc, "diagnostics", None)
raise WorkflowBpmnValidationError(
tuple(diagnostics)
if diagnostics
else (
BpmnDiagnostic(
severity="error",
code="bpmn.invalid",
message=str(exc),
),
)
) from exc
effective_graph = _validated_graph(effective_graph)
xml = export_bpmn_graph(effective_graph)
executable = not any(
item.severity == "error"
for item in runtime_diagnostics(effective_graph)
)
if executable:
try:
runtime_graph = materialize_runtime_graph(effective_graph)
except BpmnGraphError:
executable = False
else:
executable = not any(
item.severity == "error"
for item in validate_workflow_graph(runtime_graph)
)
return effective_graph, _BpmnArtifact(
xml=xml,
content_hash=hashlib.sha256(xml.encode("utf-8")).hexdigest(),
adapter_id=NATIVE_BPMN_ADAPTER_ID,
adapter_version=NATIVE_BPMN_ADAPTER_VERSION,
runtime_kind="native_graph",
executable=executable,
)
def _validate_bpmn_activation(
revision: WorkflowDefinitionRevision,
) -> None:
try:
runtime_graph = materialize_runtime_graph(
WorkflowGraph.model_validate(revision.graph)
)
except BpmnGraphError as exc:
diagnostics = getattr(exc, "diagnostics", None)
raise WorkflowBpmnValidationError(
tuple(diagnostics)
if diagnostics
else (
BpmnDiagnostic(
severity="error",
code="bpmn.activation_invalid",
message=str(exc),
),
)
) from exc
runtime_validation = validate_workflow_graph(runtime_graph)
if any(item.severity == "error" for item in runtime_validation):
raise WorkflowValidationError(runtime_validation)
def _validated_graph(graph: WorkflowGraph) -> WorkflowGraph:
try:
graph = canonical_bpmn_graph(graph)
except BpmnGraphError as exc:
raise WorkflowBpmnValidationError(exc.diagnostics) from exc
diagnostics = validate_workflow_graph(graph)
if any(item.severity == "error" for item in diagnostics):
raise WorkflowValidationError(diagnostics)
return graph
def _canonical_graph(graph: WorkflowGraph) -> dict[str, object]:
return graph.model_dump(mode="json")
def _content_hash(
graph: WorkflowGraph,
*,
bpmn: _BpmnArtifact | None = None,
execution_mode: str = "hybrid",
view_id: str | None = None,
view_revision_id: str | None = None,
) -> str:
content: dict[str, object] = {
"graph": _canonical_graph(graph),
"execution": {
"mode": execution_mode,
"view_id": view_id,
"view_revision_id": view_revision_id,
},
}
if bpmn is not None:
content["bpmn"] = {
"xml": bpmn.xml,
"adapter_id": bpmn.adapter_id,
"adapter_version": bpmn.adapter_version,
}
encoded = json.dumps(content, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
def _revision_matches(
revision: WorkflowDefinitionRevision,
*,
graph: WorkflowGraph,
bpmn: _BpmnArtifact | None,
execution_mode: str,
view_id: str | None,
view_revision_id: str | None,
) -> bool:
return (
revision.graph == _canonical_graph(graph)
and revision.execution_mode == execution_mode
and revision.view_id == view_id
and revision.view_revision_id == view_revision_id
and revision.bpmn_xml == (bpmn.xml if bpmn else None)
and revision.bpmn_adapter_id
== (bpmn.adapter_id if bpmn else None)
and revision.bpmn_adapter_version
== (bpmn.adapter_version if bpmn else None)
)
def _validate_execution_mode_activation(
revision: WorkflowDefinitionRevision,
) -> None:
if revision.execution_mode != "automated":
return
try:
graph = materialize_runtime_graph(
WorkflowGraph.model_validate(revision.graph)
)
except BpmnGraphError as exc:
raise WorkflowBpmnValidationError(exc.diagnostics) from exc
human_nodes = [
node.id
for node in graph.nodes
if node.type in {"workflow.activity", "workflow.review"}
or (
node.type == "workflow.wait"
and str(node.config.get("mode") or "manual") == "manual"
)
or (
node.type == "workflow.capability"
and str(node.config.get("failure_policy") or "manual") == "manual"
)
or (
node.type == "workflow.dataflow"
and (
str(node.config.get("warning_policy") or "review") == "review"
or str(node.config.get("failure_policy") or "manual")
== "manual"
)
)
]
if human_nodes:
raise WorkflowConflictError(
"Automated workflows cannot activate with human handoff paths: "
+ ", ".join(sorted(human_nodes))
)
def _available_key(
session: Session,
*,
scope_key: str,
requested: str | None,
name: str,
) -> str:
base = _slug(requested or name)
candidate = base
suffix = 2
while session.scalar(
select(WorkflowDefinition.id).where(
WorkflowDefinition.scope_key == scope_key,
WorkflowDefinition.definition_key == candidate,
)
):
candidate = f"{base[: max(1, 120 - len(str(suffix)) - 1)]}-{suffix}"
suffix += 1
return candidate
def _slug(value: str) -> str:
normalized = unicodedata.normalize("NFKD", value)
ascii_value = normalized.encode("ascii", "ignore").decode("ascii").lower()
cleaned = re.sub(r"[^a-z0-9]+", "-", ascii_value).strip("-")
return (cleaned or "workflow")[:120]
def _clean_optional(value: str | None) -> str | None:
if value is None:
return None
cleaned = value.strip()
return cleaned or None
def _ancestor_governance_limits(
provenance: Mapping[str, object],
) -> dict[str, bool]:
raw = provenance.get("source_effective_limits")
limits = raw if isinstance(raw, Mapping) else {}
return {
key: value if isinstance((value := limits.get(key)), bool) else True
for key in (
"inherit_to_lower_scopes",
"allow_start",
"allow_reuse",
"allow_automation",
)
}
def _effective_governance_limits(
definition: WorkflowDefinition,
*,
decision_details: Mapping[str, object] | None = None,
) -> dict[str, bool]:
ancestor = _ancestor_governance_limits(
definition.derivation_provenance
)
effective = {
"inherit_to_lower_scopes": (
definition.inherit_to_lower_scopes
and ancestor["inherit_to_lower_scopes"]
),
"allow_start": (
definition.allow_start and ancestor["allow_start"]
),
"allow_reuse": (
definition.allow_reuse and ancestor["allow_reuse"]
),
"allow_automation": (
definition.allow_automation
and ancestor["allow_automation"]
),
}
policy_limits = (
decision_details.get("effective_limits")
if decision_details is not None
else None
)
if isinstance(policy_limits, Mapping):
policy_key_by_local_key = {
"inherit_to_lower_scopes": "inherit_to_lower_scopes",
"allow_start": "allow_run",
"allow_reuse": "allow_reuse",
"allow_automation": "allow_automation",
}
for local_key, policy_key in policy_key_by_local_key.items():
value = policy_limits.get(policy_key)
if isinstance(value, bool):
effective[local_key] = effective[local_key] and value
return effective
__all__ = [
"WorkflowBpmnValidationError",
"WorkflowConflictError",
"WorkflowError",
"WorkflowNotFoundError",
"WorkflowValidationError",
"activate_definition",
"archive_definition",
"create_definition",
"derive_definition",
"definition_response",
"delete_definition",
"get_definition",
"get_definition_revision",
"list_definition_revisions",
"list_definitions",
"revision_response",
"revision_bpmn_summary",
"update_definition",
]
from govoplan_workflow_engine.backend.service import * # noqa: F401,F403
+2 -261
View File
@@ -1,262 +1,3 @@
from __future__ import annotations
"""Compatibility facade for the extracted Workflow Engine backend."""
from govoplan_core.core.definition_graphs import (
DefinitionDiagnostic,
DefinitionEdge,
DefinitionNode,
validate_definition_graph,
)
from govoplan_workflow.backend.node_library import (
WORKFLOW_GRAPH_LIBRARY,
WORKFLOW_NODE_TYPES_BY_ID,
)
from govoplan_workflow.backend.schemas import WorkflowGraph
def validate_workflow_graph(graph: WorkflowGraph) -> tuple[DefinitionDiagnostic, ...]:
diagnostics = list(
validate_definition_graph(
WORKFLOW_GRAPH_LIBRARY,
nodes=tuple(DefinitionNode(id=node.id, type=node.type) for node in graph.nodes),
edges=tuple(
DefinitionEdge(
id=edge.id,
source=edge.source,
target=edge.target,
source_port=edge.source_port,
target_port=edge.target_port,
)
for edge in graph.edges
),
)
)
outgoing = {node.id: 0 for node in graph.nodes}
for edge in graph.edges:
if edge.type == "bpmn.sequenceFlow" and edge.source in outgoing:
outgoing[edge.source] += 1
for node in graph.nodes:
definition = WORKFLOW_NODE_TYPES_BY_ID.get(node.type)
if definition is None:
continue
for field in definition.config_fields:
if node.type.startswith("bpmn."):
continue
if field.required and _empty(node.config.get(field.id)):
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="node.config_required",
message=f"{definition.label} requires {field.label.lower()}.",
node_id=node.id,
field=field.id,
)
)
if _requires_outgoing(node.type) and outgoing[node.id] == 0:
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="node.outgoing_required",
message=f"{definition.label} must lead to another workflow step.",
node_id=node.id,
)
)
diagnostics.extend(_bpmn_semantic_diagnostics(graph))
diagnostics.extend(_legacy_semantic_diagnostics(graph))
return _deduplicate(diagnostics)
def _requires_outgoing(node_type: str) -> bool:
if not node_type.startswith("bpmn."):
return bool(
WORKFLOW_NODE_TYPES_BY_ID.get(node_type)
and WORKFLOW_NODE_TYPES_BY_ID[node_type].output_ports
)
return False
def _bpmn_semantic_diagnostics(
graph: WorkflowGraph,
) -> list[DefinitionDiagnostic]:
if not graph.nodes or not any(
node.type.startswith("bpmn.") for node in graph.nodes
):
return []
diagnostics: list[DefinitionDiagnostic] = []
if any(not node.type.startswith("bpmn.") for node in graph.nodes):
diagnostics.append(
DefinitionDiagnostic(
severity="warning",
code="graph.mixed_notation",
message=(
"A Workflow revision cannot mix canonical BPMN and "
"legacy Workflow nodes."
),
)
)
return diagnostics
flow_nodes = {
node.id
for node in graph.nodes
if node.type
not in {
"bpmn.participant",
"bpmn.lane",
"bpmn.dataObjectReference",
"bpmn.dataStoreReference",
"bpmn.textAnnotation",
"bpmn.group",
"bpmn.conversation",
"bpmn.callConversation",
"bpmn.subConversation",
}
}
starts = [
node for node in graph.nodes if node.type == "bpmn.startEvent"
]
ends = [node for node in graph.nodes if node.type == "bpmn.endEvent"]
if not starts:
diagnostics.append(
DefinitionDiagnostic(
severity="warning",
code="bpmn.start_event_missing",
message="A Workflow process needs at least one BPMN start event.",
)
)
if not ends:
diagnostics.append(
DefinitionDiagnostic(
severity="warning",
code="bpmn.end_event_missing",
message="A Workflow process needs at least one BPMN end event.",
)
)
node_by_id = {node.id: node for node in graph.nodes}
default_flows_by_source: dict[str, list[str]] = {}
for edge in graph.edges:
source = node_by_id.get(edge.source)
target = node_by_id.get(edge.target)
if source is None or target is None:
continue
if edge.type == "bpmn.sequenceFlow" and (
edge.source not in flow_nodes or edge.target not in flow_nodes
):
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="bpmn.sequence_flow_endpoint",
message=(
"BPMN sequence flows may only connect process flow "
"nodes. Use an association or data association here."
),
node_id=edge.source,
)
)
if edge.type == "bpmn.messageFlow":
same_process = (
source.process_id
and target.process_id
and source.process_id == target.process_id
)
if same_process:
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="bpmn.message_flow_same_process",
message=(
"BPMN message flows connect different participants; "
"use a sequence flow inside one process."
),
node_id=edge.source,
)
)
if edge.config.get("default") is True:
if edge.type != "bpmn.sequenceFlow":
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="bpmn.default_flow_type",
message="Only a BPMN sequence flow can be a default flow.",
node_id=edge.source,
)
)
else:
default_flows_by_source.setdefault(edge.source, []).append(edge.id)
for source_id, edge_ids in default_flows_by_source.items():
if len(edge_ids) > 1:
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="bpmn.multiple_default_flows",
message="A BPMN flow node can have at most one default flow.",
node_id=source_id,
)
)
for node in graph.nodes:
if node.type != "bpmn.boundaryEvent":
continue
attached_to = str(node.config.get("attached_to_ref") or "")
if attached_to not in node_by_id:
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="bpmn.boundary_attachment",
message="A boundary event must reference an activity in this graph.",
node_id=node.id,
field="attached_to_ref",
)
)
return diagnostics
def _legacy_semantic_diagnostics(
graph: WorkflowGraph,
) -> list[DefinitionDiagnostic]:
if not graph.nodes or any(
node.type.startswith("bpmn.") for node in graph.nodes
):
return []
diagnostics: list[DefinitionDiagnostic] = []
starts = [
node for node in graph.nodes if node.type.startswith("workflow.start.")
]
outcomes = [
node for node in graph.nodes if node.type.startswith("workflow.end.")
]
if len(starts) != 1:
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="graph.trigger_count",
message="A definition requires exactly one start node.",
)
)
if not outcomes:
diagnostics.append(
DefinitionDiagnostic(
severity="error",
code="graph.outcome_count",
message="A definition requires at least one outcome node.",
)
)
return diagnostics
def _empty(value: object) -> bool:
return value is None or value == "" or value == [] or value == {}
def _deduplicate(
diagnostics: list[DefinitionDiagnostic],
) -> tuple[DefinitionDiagnostic, ...]:
seen: set[tuple[str, str | None, str | None]] = set()
result: list[DefinitionDiagnostic] = []
for item in diagnostics:
key = (item.code, item.node_id, item.field)
if key in seen:
continue
seen.add(key)
result.append(item)
return tuple(result)
__all__ = ["validate_workflow_graph"]
from govoplan_workflow_engine.backend.validation import * # noqa: F401,F403