GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 19 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add workflow engine contribution contracts

Browse Source
This commit is contained in:
zemion
2026-07-31 16:58:57 +02:00
parent 5211e07d0b
commit 1884274f8d
11 changed files with 301 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/ACTION_EFFECT_AUTOMATION_LAYER.md
+2 -2
View File
@@ -5,7 +5,7 @@ only linear screen flows. Workflows, schedules, imports, connectors, policies,
and external events all need to request governed actions without bypassing the
same safety rules that apply to human users.
The first implementation should live in `govoplan-workflow` and core contracts.
The first implementation lives in `govoplan-workflow-engine` and Core contracts.
Create a separate `govoplan-automation` module only if action planning,
schedulers, rule execution, or cross-module automation become too broad for
workflow ownership.
@@ -122,7 +122,7 @@ recovery decisions, and workflow advancement.
## Boundary
Core may own stable DTOs, registry contracts, and generic audit/event hooks.
`govoplan-workflow` should own the first runner because workflow is the first
`govoplan-workflow-engine` owns the first runner because workflow is the first
module that coordinates cross-module process actions.
Domain modules own their own action providers. For example, templates own
docs/DEPLOYMENT_OPERATOR_GUIDE.md
+1 -1
View File
@@ -157,7 +157,7 @@ release evidence.
| --- | --- | --- |
| `REDIS_URL` | `redis://redis:6379/0` | Celery broker/result backend when async workers are enabled. |
| `CELERY_ENABLED` | `false` | Local/dev can send synchronously. Production campaign delivery should run workers and set this to `true`. |
| `CELERY_QUEUES` | `send_email,append_sent,notifications,calendar,dataflow,events,default` | Queue list expected by worker/process manager definitions. The `events` queue drains transactional platform events; `dataflow` drains trigger deliveries and schedules. |
| `CELERY_QUEUES` | `send_email,append_sent,notifications,calendar,dataflow,workflow,events,default` | Queue list expected by worker/process manager definitions. The `events` queue drains transactional platform events; `dataflow` drains trigger deliveries and schedules; `workflow` reconciles Workflow Engine instances and module standards. |
| `PLATFORM_EVENT_OUTBOX_MAX_ATTEMPTS` | `8` | Failed durable consumer deliveries are quarantined after this many attempts. |
| `PLATFORM_EVENT_OUTBOX_TERMINAL_RETENTION_DAYS` | `90` | Successful event envelopes older than this are removed by the daily retention task. Quarantined evidence is retained. |
docs/MODULE_ARCHITECTURE.md
+11 -1
View File
@@ -148,6 +148,16 @@ Other stable runtime capabilities currently include:
- `calendar.outbox` and `calendar.scheduling`
- `poll.scheduling`
- `notifications.dispatch`
- `workflow.definitionContributions` and `workflow.runtimeWorker`
Modules contribute reusable process baselines through
`ModuleManifest.workflow_definitions`. Each contribution pins its origin module
and version, stable key, schema and content hash, native graph/BPMN content,
governance ceilings, execution mode, and required capabilities/interfaces.
`govoplan-workflow-engine` reconciles these declarations idempotently. A module
upgrade appends a baseline revision without replacing the active revision or
mutating a local override; the optional `govoplan-workflow` package supplies
the comparison, derivation, and reset UI.
### Named Interface Contracts
@@ -889,7 +899,7 @@ First slice:
- `govoplan-files` owns file-backed governed locations and uploaded/stored file
evidence.
- `govoplan-reporting` owns report/data views and scheduled outputs.
- `govoplan-workflow` owns process state, approvals, scheduling of process
- `govoplan-workflow-engine` owns process state, approvals, scheduling of process
steps, and human review.
Future `govoplan-datasources` is justified when GovOPlaN needs a broad source
docs/RELEASE_DEPENDENCIES.md
+2 -1
View File
@@ -151,7 +151,8 @@ Current tag-only module repositories:
- `govoplan-search`
- `govoplan-tasks`
- `govoplan-templates`
- `govoplan-workflow`
- `govoplan-workflow-engine` (headless definitions, migrations, and execution)
- `govoplan-workflow` (optional authoring and inspection WebUI)
- `govoplan-xoev`
- `govoplan-xrechnung`
- `govoplan-xta-osci`
src/govoplan_core/core/install_config.py
+2 -2
View File
@@ -373,7 +373,7 @@ ENABLED_MODULES=tenancy,organizations,identity,access,admin,dashboard,policy,aud
CELERY_ENABLED=true
REDIS_URL=redis://127.0.0.1:6379/0
CELERY_QUEUES=send_email,append_sent,notifications,calendar,dataflow,events,default
CELERY_QUEUES=send_email,append_sent,notifications,calendar,dataflow,workflow,events,default
CALENDAR_OUTBOX_TERMINAL_RETENTION_DAYS=90
PLATFORM_EVENT_OUTBOX_MAX_ATTEMPTS=8
PLATFORM_EVENT_OUTBOX_TERMINAL_RETENTION_DAYS=90
@@ -446,7 +446,7 @@ DATABASE_URL=postgresql+psycopg://govoplan:govoplan-dev@127.0.0.1:55433/govoplan
GOVOPLAN_DATABASE_URL_PGTOOLS=postgresql://govoplan:govoplan-dev@127.0.0.1:55433/govoplan
REDIS_URL=redis://127.0.0.1:56379/0
CELERY_ENABLED=true
CELERY_QUEUES=send_email,append_sent,notifications,calendar,dataflow,events,default
CELERY_QUEUES=send_email,append_sent,notifications,calendar,dataflow,workflow,events,default
CALENDAR_OUTBOX_TERMINAL_RETENTION_DAYS=90
PLATFORM_EVENT_OUTBOX_MAX_ATTEMPTS=8
PLATFORM_EVENT_OUTBOX_TERMINAL_RETENTION_DAYS=90
src/govoplan_core/core/modules.py
+6
View File
@@ -13,6 +13,7 @@ if TYPE_CHECKING:
SearchProviderRegistration,
SearchSourceProviderRegistration,
)
from govoplan_core.core.workflows import WorkflowDefinitionContribution
SUPPORTED_MANIFEST_CONTRACT_VERSION = "1"
@@ -440,3 +441,8 @@ class ModuleManifest:
...,
] = ()
documentation_sources: tuple[DocumentationSourceDefinition, ...] = ()
# A renamed or extracted module may continue to own an established
# permission namespace. This keeps persisted grants stable while the
# runtime module ID changes.
permission_namespace: str | None = None
workflow_definitions: tuple["WorkflowDefinitionContribution", ...] = ()
src/govoplan_core/core/registry.py
+66 -2
View File
@@ -43,6 +43,9 @@ from govoplan_core.core.views import (
route_view_surface_id,
validate_view_surface_id,
)
from govoplan_core.core.workflows import (
workflow_definition_contribution_hash,
)
_MODULE_ID_RE = re.compile(r"^[a-z][a-z0-9_]*$")
_NPM_PACKAGE_RE = re.compile(r"^(?:@[a-z0-9][a-z0-9_.-]*/)?[a-z0-9][a-z0-9_.-]*$")
@@ -540,8 +543,12 @@ def _validate_manifest_permission(
permission: PermissionDefinition,
seen_permissions: Mapping[str, PermissionDefinition],
) -> None:
if permission.module_id != manifest.id:
raise RegistryError(f"Permission {permission.scope!r} has mismatched module id {permission.module_id!r}")
permission_namespace = manifest.permission_namespace or manifest.id
if permission.module_id != permission_namespace:
raise RegistryError(
f"Permission {permission.scope!r} has mismatched module id "
f"{permission.module_id!r}; expected {permission_namespace!r}"
)
if not _SCOPE_RE.match(permission.scope):
raise RegistryError(f"Permission scope must be <module>:<resource>:<action>: {permission.scope!r}")
expected_prefix = f"{permission.module_id}:{permission.resource}:"
@@ -611,6 +618,55 @@ def _validate_manifest_shape(manifest: ModuleManifest) -> None:
f"Module {manifest.id!r} documentation topic {topic.id!r}: {issue}"
)
_validate_documentation_extensions(manifest)
_validate_workflow_definition_contributions(manifest)
def _validate_workflow_definition_contributions(
manifest: ModuleManifest,
) -> None:
seen_keys: set[str] = set()
for contribution in manifest.workflow_definitions:
if contribution.origin_module_id != manifest.id:
raise RegistryError(
f"Workflow contribution {contribution.definition_key!r} in "
f"module {manifest.id!r} declares origin module "
f"{contribution.origin_module_id!r}"
)
if contribution.origin_module_version != manifest.version:
raise RegistryError(
f"Workflow contribution {contribution.definition_key!r} in "
f"module {manifest.id!r} declares origin version "
f"{contribution.origin_module_version!r}, expected "
f"{manifest.version!r}"
)
if not re.match(
r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,119}$",
contribution.definition_key,
):
raise RegistryError(
f"Workflow contribution key is invalid: "
f"{contribution.definition_key!r}"
)
if contribution.definition_key in seen_keys:
raise RegistryError(
f"Module {manifest.id!r} declares duplicate Workflow "
f"contribution {contribution.definition_key!r}"
)
seen_keys.add(contribution.definition_key)
if contribution.scope_type not in {"system", "tenant"}:
raise RegistryError(
f"Workflow contribution {contribution.definition_key!r} has "
f"unsupported scope {contribution.scope_type!r}"
)
expected_hash = workflow_definition_contribution_hash(contribution)
if (
contribution.content_hash is not None
and contribution.content_hash != expected_hash
):
raise RegistryError(
f"Workflow contribution {contribution.definition_key!r} has "
"a content hash that does not match its canonical content"
)
def _validate_documentation_extensions(manifest: ModuleManifest) -> None:
@@ -678,6 +734,14 @@ def _validate_manifest_identity(manifest: ModuleManifest) -> None:
raise RegistryError(f"Module {manifest.id!r} must declare a non-empty name")
if not manifest.version.strip():
raise RegistryError(f"Module {manifest.id!r} must declare a non-empty version")
if (
manifest.permission_namespace is not None
and not _MODULE_ID_RE.match(manifest.permission_namespace)
):
raise RegistryError(
f"Module {manifest.id!r} has invalid permission namespace "
f"{manifest.permission_namespace!r}"
)
if manifest.compatibility.manifest_contract_version != SUPPORTED_MANIFEST_CONTRACT_VERSION:
raise RegistryError(
f"Module {manifest.id!r} uses unsupported manifest contract version "
src/govoplan_core/core/workflows.py
+113 -1
View File
@@ -1,11 +1,86 @@
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from datetime import datetime
import hashlib
import json
from typing import Protocol, runtime_checkable
CAPABILITY_WORKFLOW_RUNTIME_WORKER = "workflow.runtimeWorker"
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS = (
"workflow.definitionContributions"
)
@dataclass(frozen=True, slots=True)
class WorkflowDefinitionContribution:
"""Versioned, module-owned Workflow Engine baseline definition."""
origin_module_id: str
origin_module_version: str
definition_key: str
name: str
graph: Mapping[str, object]
contribution_schema_version: str = "1"
description: str | None = None
definition_kind: str = "flow"
scope_type: str = "system"
inherit_to_lower_scopes: bool = True
allow_start: bool = True
allow_reuse: bool = True
allow_automation: bool = False
execution_mode: str = "hybrid"
view_id: str | None = None
view_revision_id: str | None = None
bpmn_xml: str | None = None
bpmn_adapter_id: str = "govoplan.native.bpmn"
bpmn_adapter_version: str | None = None
activate_on_install: bool = True
required_capabilities: tuple[str, ...] = ()
required_interfaces: tuple[str, ...] = ()
metadata: Mapping[str, object] = field(default_factory=dict)
policy_metadata: Mapping[str, object] = field(default_factory=dict)
content_hash: str | None = None
def workflow_definition_contribution_hash(
contribution: WorkflowDefinitionContribution,
) -> str:
"""Return the stable content hash, excluding package release provenance."""
payload = {
"schema_version": contribution.contribution_schema_version,
"definition_key": contribution.definition_key,
"name": contribution.name,
"description": contribution.description,
"graph": contribution.graph,
"definition_kind": contribution.definition_kind,
"scope_type": contribution.scope_type,
"inherit_to_lower_scopes": contribution.inherit_to_lower_scopes,
"allow_start": contribution.allow_start,
"allow_reuse": contribution.allow_reuse,
"allow_automation": contribution.allow_automation,
"execution_mode": contribution.execution_mode,
"view_id": contribution.view_id,
"view_revision_id": contribution.view_revision_id,
"bpmn_xml": contribution.bpmn_xml,
"bpmn_adapter_id": contribution.bpmn_adapter_id,
"bpmn_adapter_version": contribution.bpmn_adapter_version,
"activate_on_install": contribution.activate_on_install,
"required_capabilities": contribution.required_capabilities,
"required_interfaces": contribution.required_interfaces,
"metadata": contribution.metadata,
"policy_metadata": contribution.policy_metadata,
}
encoded = json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
)
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
@runtime_checkable
@@ -20,6 +95,17 @@ class WorkflowRuntimeWorker(Protocol):
...
@runtime_checkable
class WorkflowDefinitionContributionProvider(Protocol):
def reconcile(
self,
session: object,
*,
tenant_ids: Sequence[str] = (),
) -> Mapping[str, object]:
...
def workflow_runtime_worker(
registry: object | None,
) -> WorkflowRuntimeWorker | None:
@@ -37,8 +123,34 @@ def workflow_runtime_worker(
)
def workflow_definition_contribution_provider(
registry: object | None,
) -> WorkflowDefinitionContributionProvider | None:
if (
registry is None
or not hasattr(registry, "has_capability")
or not registry.has_capability(
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS
)
):
return None
capability = registry.capability(
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS
)
return (
capability
if isinstance(capability, WorkflowDefinitionContributionProvider)
else None
)
__all__ = [
"CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS",
"CAPABILITY_WORKFLOW_RUNTIME_WORKER",
"WorkflowDefinitionContribution",
"WorkflowDefinitionContributionProvider",
"WorkflowRuntimeWorker",
"workflow_definition_contribution_hash",
"workflow_definition_contribution_provider",
"workflow_runtime_worker",
]
src/govoplan_core/settings.py
+2 -2
View File
@@ -49,7 +49,7 @@ class Settings(BaseSettings):
default=(
"tenancy,organizations,identity,idm,access,admin,dashboard,policy,"
"audit,campaigns,files,mail,calendar,poll,scheduling,connectors,"
"datasources,dataflow,workflow,views,search,risk_compliance,"
"datasources,dataflow,workflow_engine,workflow,views,search,risk_compliance,"
"postbox,notifications,docs,ops"
),
alias="ENABLED_MODULES",
@@ -156,7 +156,7 @@ class Settings(BaseSettings):
celery_queues: str = Field(
default=(
"send_email,append_sent,notifications,calendar,"
"dataflow,events,default"
"dataflow,workflow,events,default"
),
alias="CELERY_QUEUES",
)
tests/test_module_system.py
+51 -1
View File
@@ -101,7 +101,7 @@ from govoplan_core.core.module_package_catalog import (
)
from govoplan_core.core.modules import FrontendModule, FrontendRoute, MigrationRetirementPlan, ModuleCompatibility, ModuleMigrationTask, ModuleMigrationTaskContext, ModuleMigrationTaskResult, ModuleUninstallGuardResult, PublicFrontendRoute
from govoplan_core.core.module_guards import drop_table_retirement_provider
from govoplan_core.core.modules import MigrationSpec, ModuleInterfaceProvider, ModuleInterfaceRequirement, ModuleManifest, RoleTemplate
from govoplan_core.core.modules import MigrationSpec, ModuleInterfaceProvider, ModuleInterfaceRequirement, ModuleManifest, PermissionDefinition, RoleTemplate
from govoplan_core.core.registry import PlatformRegistry, RegistryError
from govoplan_core.admin.models import SystemSettings
from govoplan_core.db.base import Base
@@ -363,6 +363,56 @@ class ModuleSystemTests(unittest.TestCase):
with self.assertRaisesRegex(RegistryError, "unsupported manifest contract version"):
registry.validate()
def test_manifest_can_preserve_a_legacy_permission_namespace(self) -> None:
registry = PlatformRegistry()
registry.register(
ModuleManifest(
id="workflow_engine",
name="Workflow Engine",
version="test",
permission_namespace="workflow",
permissions=(
PermissionDefinition(
scope="workflow:definition:read",
label="Read workflows",
description="Read workflow definitions.",
category="Workflow",
level="tenant",
module_id="workflow",
resource="definition",
action="read",
),
),
)
)
registry.validate()
def test_manifest_rejects_an_undeclared_permission_namespace(self) -> None:
registry = PlatformRegistry()
registry.register(
ModuleManifest(
id="workflow_engine",
name="Workflow Engine",
version="test",
permissions=(
PermissionDefinition(
scope="workflow:definition:read",
label="Read workflows",
description="Read workflow definitions.",
category="Workflow",
level="tenant",
module_id="workflow",
resource="definition",
action="read",
),
),
)
)
with self.assertRaisesRegex(RegistryError, "expected 'workflow_engine'"):
registry.validate()
def test_default_authenticated_role_templates_are_managed_tenant_roles(self) -> None:
for template, message in (
(
tests/test_workflow_runtime_worker.py
+45
View File
@@ -10,8 +10,13 @@ from govoplan_core.celery_app import (
from govoplan_core.core.modules import ModuleContext, ModuleManifest
from govoplan_core.core.registry import PlatformRegistry
from govoplan_core.core.workflows import (
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS,
CAPABILITY_WORKFLOW_RUNTIME_WORKER,
WorkflowDefinitionContribution,
WorkflowDefinitionContributionProvider,
WorkflowRuntimeWorker,
workflow_definition_contribution_hash,
workflow_definition_contribution_provider,
workflow_runtime_worker,
)
@@ -21,7 +26,47 @@ class _Worker:
return {"inspected": 0, "advanced": 0}
class _ContributionProvider:
def reconcile(self, session, *, tenant_ids=()):
return {"created": 0, "tenant_ids": tuple(tenant_ids)}
class WorkflowRuntimeWorkerTests(unittest.TestCase):
def test_contribution_contract_hash_and_provider_are_stable(self) -> None:
contribution = WorkflowDefinitionContribution(
origin_module_id="example",
origin_module_version="1.0.0",
definition_key="approval",
name="Approval",
graph={"schema_version": 1, "nodes": [], "edges": []},
)
self.assertEqual(
workflow_definition_contribution_hash(contribution),
workflow_definition_contribution_hash(contribution),
)
provider = _ContributionProvider()
self.assertIsInstance(provider, WorkflowDefinitionContributionProvider)
registry = PlatformRegistry()
registry.register(
ModuleManifest(
id="workflow_engine_test",
name="Workflow Engine test",
version="test",
capability_factories={
CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS: (
lambda _context: provider
),
},
)
)
registry.configure_capability_context(
ModuleContext(registry=registry, settings=object())
)
self.assertIs(
provider,
workflow_definition_contribution_provider(registry),
)
def test_contract_is_runtime_checkable_and_resolved(self) -> None:
worker = _Worker()
self.assertIsInstance(worker, WorkflowRuntimeWorker)