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

feat: add governed service workflow launcher

Browse Source
This commit is contained in:
zemion
2026-08-01 17:48:41 +02:00
parent 43c1f3fb72
commit 55f98d1b65
7 changed files with 272 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_workflow_engine/backend/manifest.py
+38 -1
View File
@@ -14,6 +14,7 @@ from govoplan_core.core.module_guards import (
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationTopic,
MigrationSpec,
ModuleContext,
@@ -23,6 +24,7 @@ from govoplan_core.core.modules import (
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.policy import (
CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
)
@@ -41,6 +43,10 @@ from govoplan_workflow_engine.backend.db import models as workflow_models
from govoplan_workflow_engine.backend.configuration_provider import (
WORKFLOW_CONFIGURATION_CAPABILITY,
)
from govoplan_workflow_engine.backend.service_launcher import (
CAPABILITY_WORKFLOW_SERVICE_LAUNCHER,
WorkflowServiceLauncher,
)
MODULE_ID = "workflow_engine"
@@ -164,6 +170,10 @@ def _orchestration_provider(context: ModuleContext):
return SqlWorkflowOrchestrationProvider(registry=context.registry)
def _service_launcher(context: ModuleContext) -> WorkflowServiceLauncher:
return WorkflowServiceLauncher(registry=context.registry)
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
@@ -209,6 +219,10 @@ manifest = ModuleManifest(
),
ModuleInterfaceProvider(name="workflow.runtime_worker", version=MODULE_VERSION),
ModuleInterfaceProvider(name="workflow.bpmn_interchange", version="1.0.0"),
ModuleInterfaceProvider(
name=CAPABILITY_WORKFLOW_SERVICE_LAUNCHER,
version="0.1.0",
),
ModuleInterfaceProvider(
name="workflow.bpmn_execution_adapters",
version="1.0.0",
@@ -262,6 +276,14 @@ manifest = ModuleManifest(
CAPABILITY_WORKFLOW_RUNTIME_WORKER: _runtime_worker,
CAPABILITY_WORKFLOW_ORCHESTRATION: _orchestration_provider,
WORKFLOW_CONFIGURATION_CAPABILITY: _configuration_provider,
CAPABILITY_WORKFLOW_SERVICE_LAUNCHER: _service_launcher,
},
capability_documentation={
CAPABILITY_WORKFLOW_SERVICE_LAUNCHER: CapabilityDocumentation(
label="Workflow service launcher",
summary="Starts an authorized active Workflow from an exact available Service revision.",
contract_version="0.1.0",
),
},
migration_spec=MigrationSpec(
module_id=MODULE_ID,
@@ -302,7 +324,9 @@ manifest = ModuleManifest(
"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."
"revisions; activation pins the exact revision used by future instances. "
"The optional service-launch capability starts an authorized active "
"revision from an exact Portal Service binding and records that provenance."
),
layer="available",
documentation_types=("admin", "user"),
@@ -338,6 +362,19 @@ manifest = ModuleManifest(
order=77,
),
),
architecture=declared_module_architecture(
layer="human_work_procedure",
kind="runtime",
maturity="vertical_slice",
documentation_ref="docs/ENGINE_EDITOR_SPLIT.md",
test_ref="tests/test_instance_service.py",
known_limits=("Execution adapters support declared conformance profiles but do not cover every editable BPMN semantic.",),
owned_concepts=("workflow definition", "workflow revision", "workflow instance", "work transition", "execution adapter binding"),
non_owned_concepts=("visual editor", "domain action", "notification", "dataflow run"),
recovery_docs=("docs/CONCEPT.md",),
security_docs=("docs/CONCEPT.md",),
operations_docs=("README.md",),
),
)
src/govoplan_workflow_engine/backend/service_launcher.py
+142
View File
@@ -0,0 +1,142 @@
from __future__ import annotations
import hashlib
from urllib.parse import quote
from sqlalchemy import case, or_, select
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.institutional import (
InstitutionalContextError,
InstitutionalReference,
ServiceDefinition,
ServiceLaunchRequest,
ServiceLaunchResult,
)
from govoplan_workflow_engine.backend.db.models import WorkflowDefinition
from govoplan_workflow_engine.backend.instance_service import start_instance
from govoplan_workflow_engine.backend.schemas import WorkflowInstanceStartRequest
CAPABILITY_WORKFLOW_SERVICE_LAUNCHER = "workflow_engine.service_launcher"
class WorkflowServiceLauncher:
def __init__(self, registry: object | None = None) -> None:
self._registry = registry
def launch_service(
self,
session: object,
principal: object,
*,
definition: ServiceDefinition,
request: ServiceLaunchRequest,
) -> ServiceLaunchResult:
if not isinstance(session, Session) or not isinstance(principal, ApiPrincipal):
raise InstitutionalContextError(
"Workflow service launch requires a database session and API principal."
)
if request.service_ref != definition.reference:
raise InstitutionalContextError(
"Workflow service launch must use the requested exact Service revision."
)
if (
request.binding.kind != "workflow"
or request.binding not in definition.bindings
):
raise InstitutionalContextError(
"Workflow service launch requires a workflow binding from the Service definition."
)
workflow = _workflow_definition(
session,
tenant_id=definition.reference.tenant_id,
reference=request.binding.reference,
)
correlation_hash = hashlib.sha256(
(
f"{definition.reference.tenant_id}:"
f"{definition.reference.object_id}:"
f"{definition.reference.version}:"
f"{request.idempotency_key}"
).encode("utf-8")
).hexdigest()[:32]
instance, replayed = start_instance(
session,
tenant_id=definition.reference.tenant_id,
definition_id=workflow.id,
actor_id=principal.account_id,
principal=principal,
registry=self._registry,
payload=WorkflowInstanceStartRequest(
idempotency_key=request.idempotency_key,
input={
**dict(request.parameters),
"_service": definition.reference.to_dict(),
},
correlation_id=f"service:{correlation_hash}",
),
start_origin="user",
)
target_ref = InstitutionalReference(
kind="workflow",
owner_module="workflow_engine",
object_id=instance.id,
tenant_id=definition.reference.tenant_id,
version=str(workflow.active_revision or workflow.current_revision),
valid_at=request.requested_at,
)
return ServiceLaunchResult(
service_ref=definition.reference,
binding=request.binding,
state="started",
target_ref=target_ref,
href=f"/workflow?instance={quote(instance.id, safe='')}",
replayed=replayed,
metadata={
"workflow_definition_id": workflow.id,
"workflow_instance_id": instance.id,
"workflow_status": instance.status,
},
)
def _workflow_definition(
session: Session,
*,
tenant_id: str,
reference: str,
) -> WorkflowDefinition:
value = reference.removeprefix("workflow:").strip()
if not value:
raise InstitutionalContextError("Workflow Service binding is invalid.")
item = session.scalar(
select(WorkflowDefinition)
.where(
or_(
WorkflowDefinition.tenant_id == tenant_id,
WorkflowDefinition.tenant_id.is_(None),
),
WorkflowDefinition.deleted_at.is_(None),
or_(
WorkflowDefinition.id == value,
WorkflowDefinition.definition_key == value,
),
)
.order_by(
case((WorkflowDefinition.tenant_id == tenant_id, 0), else_=1),
WorkflowDefinition.updated_at.desc(),
)
)
if item is None:
raise InstitutionalContextError(
"Workflow Service binding does not resolve to an available definition."
)
return item
__all__ = [
"CAPABILITY_WORKFLOW_SERVICE_LAUNCHER",
"WorkflowServiceLauncher",
]