From 6ebb299d6ced93f5ae45cb7af7435900a00a3f69 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 31 Jul 2026 19:39:59 +0200 Subject: [PATCH] Add workflow baseline orchestration contracts --- src/govoplan_core/core/lifecycle.py | 23 +++++ src/govoplan_core/core/organizations.py | 6 +- src/govoplan_core/core/policy.py | 91 +++++++++++++++++++ src/govoplan_core/core/registry.py | 3 + src/govoplan_core/core/workflows.py | 89 +++++++++++++++++- src/govoplan_core/server/default_config.py | 6 ++ tests/test_workflow_contribution_lifecycle.py | 91 +++++++++++++++++++ 7 files changed, 304 insertions(+), 5 deletions(-) create mode 100644 tests/test_workflow_contribution_lifecycle.py diff --git a/src/govoplan_core/core/lifecycle.py b/src/govoplan_core/core/lifecycle.py index e0c2814..a8c9bbd 100644 --- a/src/govoplan_core/core/lifecycle.py +++ b/src/govoplan_core/core/lifecycle.py @@ -10,6 +10,9 @@ from govoplan_core.core.module_management import ModuleManagementError, REQUIRED from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.runtime import configure_runtime +from govoplan_core.core.workflows import ( + workflow_definition_contribution_provider, +) from govoplan_core.server.route_validation import validate_router_can_mount @@ -116,6 +119,8 @@ class ModuleLifecycleManager: if hook is not None: hook(self.context) + self.reconcile_workflow_definitions() + if self._app is not None: self._app.openapi_schema = None @@ -127,6 +132,24 @@ class ModuleLifecycleManager: migrations_applied=migrate, ) + def reconcile_workflow_definitions(self) -> Mapping[str, object]: + if not any( + manifest.workflow_definitions for manifest in self.registry.manifests() + ): + return {"skipped": True, "reason": "no_contributions"} + provider = workflow_definition_contribution_provider(self.registry) + if provider is None: + return {"skipped": True, "reason": "provider_unavailable"} + + from govoplan_core.db.session import get_database + + with get_database().session() as session: + result = provider.reconcile(session) + session.commit() + if self._app is not None: + self._app.state.govoplan_workflow_reconciliation = dict(result) + return dict(result) + def _mount_module_router(self, module_id: str) -> bool: if module_id in self._mounted_modules: return False diff --git a/src/govoplan_core/core/organizations.py b/src/govoplan_core/core/organizations.py index 4ddb6e6..abdc704 100644 --- a/src/govoplan_core/core/organizations.py +++ b/src/govoplan_core/core/organizations.py @@ -1,7 +1,7 @@ from __future__ import annotations -from collections.abc import Sequence -from dataclasses import dataclass +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field from datetime import datetime from typing import Literal, Protocol, runtime_checkable @@ -79,6 +79,7 @@ class OrganizationFunctionRef: delegable: bool = False act_in_place_allowed: bool = False status: OrganizationStatus = "active" + settings: Mapping[str, object] = field(default_factory=dict) @dataclass(frozen=True, slots=True) @@ -102,6 +103,7 @@ class OrganizationFunctionTypeRef: delegable: bool = False act_in_place_allowed: bool = False status: OrganizationStatus = "active" + settings: Mapping[str, object] = field(default_factory=dict) @dataclass(frozen=True, slots=True) diff --git a/src/govoplan_core/core/policy.py b/src/govoplan_core/core/policy.py index 3c97701..ab1838f 100644 --- a/src/govoplan_core/core/policy.py +++ b/src/govoplan_core/core/policy.py @@ -26,11 +26,25 @@ ViewGovernanceAction = Literal[ "derive", "workflow_activate", ] +FunctionAssignmentChangeKind = Literal["request", "grant"] +FunctionAssignmentGovernanceAction = Literal[ + "submit", + "approve_holder", + "approve_authority", + "accept_recipient", + "request_changes", + "respond", + "reject", + "withdraw", + "recover", + "apply", +] CAPABILITY_POLICY_PRIVACY_RETENTION = "policy.privacyRetention" CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY = "policy.schedulingParticipantPrivacy" CAPABILITY_POLICY_DEFINITION_GOVERNANCE = "policy.definitionGovernance" CAPABILITY_POLICY_VIEW_GOVERNANCE = "policy.viewGovernance" +CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE = "policy.functionAssignmentGovernance" POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ( "system", @@ -168,6 +182,83 @@ class PolicyDecision: } +@dataclass(frozen=True, slots=True) +class FunctionAssignmentGovernanceRequest: + tenant_id: str + kind: FunctionAssignmentChangeKind + action: FunctionAssignmentGovernanceAction + function_id: str + actor: PrincipalRef + candidate_identity_id: str + candidate_account_id: str | None = None + current_state: str = "draft" + function_settings: Mapping[str, Any] = field(default_factory=dict) + context: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class FunctionAssignmentGovernanceDecision: + allowed: bool + reason: str | None = None + profile: str = "unavailable" + required_steps: tuple[str, ...] = () + authority_function_id: str | None = None + evidence_required: bool = False + recipient_acceptance_required: bool = False + separation_of_duties: bool = True + quorum: int = 1 + maximum_validity_days: int | None = None + request_expiry_hours: int = 336 + source_path: tuple[PolicySourceStep, ...] = () + requirements: tuple[str, ...] = () + details: Mapping[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "allowed": self.allowed, + "reason": self.reason, + "profile": self.profile, + "required_steps": list(self.required_steps), + "authority_function_id": self.authority_function_id, + "evidence_required": self.evidence_required, + "recipient_acceptance_required": (self.recipient_acceptance_required), + "separation_of_duties": self.separation_of_duties, + "quorum": self.quorum, + "maximum_validity_days": self.maximum_validity_days, + "request_expiry_hours": self.request_expiry_hours, + "source_path": [step.to_dict() for step in self.source_path], + "requirements": list(self.requirements), + "details": dict(self.details), + } + + +@runtime_checkable +class FunctionAssignmentGovernancePolicy(Protocol): + def resolve_function_assignment_action( + self, + session: object | None = None, + *, + request: FunctionAssignmentGovernanceRequest, + ) -> FunctionAssignmentGovernanceDecision: ... + + +def function_assignment_governance_policy( + registry: object | None, +) -> FunctionAssignmentGovernancePolicy | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE) + ): + return None + capability = registry.capability(CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE) + return ( + capability + if isinstance(capability, FunctionAssignmentGovernancePolicy) + else None + ) + + @dataclass(frozen=True, slots=True) class DefinitionScopeRef: scope_type: DefinitionScopeType diff --git a/src/govoplan_core/core/registry.py b/src/govoplan_core/core/registry.py index e7fdf45..aceadfe 100644 --- a/src/govoplan_core/core/registry.py +++ b/src/govoplan_core/core/registry.py @@ -234,6 +234,9 @@ class PlatformRegistry: def has_capability(self, name: str) -> bool: return name in self._capability_factories + def capability_names(self) -> tuple[str, ...]: + return tuple(sorted(self._capability_factories)) + def capability(self, name: str) -> object | None: if name in self._capabilities: return self._capabilities[name] diff --git a/src/govoplan_core/core/workflows.py b/src/govoplan_core/core/workflows.py index 6945c76..5aa75f6 100644 --- a/src/govoplan_core/core/workflows.py +++ b/src/govoplan_core/core/workflows.py @@ -9,9 +9,8 @@ from typing import Protocol, runtime_checkable CAPABILITY_WORKFLOW_RUNTIME_WORKER = "workflow.runtimeWorker" -CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS = ( - "workflow.definitionContributions" -) +CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS = "workflow.definitionContributions" +CAPABILITY_WORKFLOW_ORCHESTRATION = "workflow.orchestration" @dataclass(frozen=True, slots=True) @@ -45,6 +44,42 @@ class WorkflowDefinitionContribution: content_hash: str | None = None +@dataclass(frozen=True, slots=True) +class WorkflowStandardStartRequest: + tenant_id: str + origin_module_id: str + definition_key: str + idempotency_key: str + input: Mapping[str, object] = field(default_factory=dict) + actor_id: str | None = None + correlation_id: str | None = None + start_origin: str = "user" + + +@dataclass(frozen=True, slots=True) +class WorkflowCurrentStepResolution: + action: str + expected_step_id: str | None = None + actor_id: str | None = None + output: Mapping[str, object] = field(default_factory=dict) + evidence: tuple[str, ...] = () + comment: str | None = None + + +@dataclass(frozen=True, slots=True) +class WorkflowInstanceRef: + id: str + tenant_id: str + definition_id: str + definition_revision_id: str + definition_revision: int + definition_hash: str + status: str + current_step_id: str | None = None + current_node_id: str | None = None + replayed: bool = False + + def workflow_definition_contribution_hash( contribution: WorkflowDefinitionContribution, ) -> str: @@ -106,6 +141,35 @@ class WorkflowDefinitionContributionProvider(Protocol): ... +@runtime_checkable +class WorkflowOrchestrationProvider(Protocol): + def start_standard( + self, + session: object, + principal: object, + *, + request: WorkflowStandardStartRequest, + ) -> WorkflowInstanceRef: ... + + def resolve_current_step( + self, + session: object, + principal: object, + *, + tenant_id: str, + instance_id: str, + resolution: WorkflowCurrentStepResolution, + ) -> WorkflowInstanceRef: ... + + def get_instance( + self, + session: object, + *, + tenant_id: str, + instance_id: str, + ) -> WorkflowInstanceRef: ... + + def workflow_runtime_worker( registry: object | None, ) -> WorkflowRuntimeWorker | None: @@ -144,13 +208,32 @@ def workflow_definition_contribution_provider( ) +def workflow_orchestration_provider( + registry: object | None, +) -> WorkflowOrchestrationProvider | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_WORKFLOW_ORCHESTRATION) + ): + return None + capability = registry.capability(CAPABILITY_WORKFLOW_ORCHESTRATION) + return capability if isinstance(capability, WorkflowOrchestrationProvider) else None + + __all__ = [ "CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS", + "CAPABILITY_WORKFLOW_ORCHESTRATION", "CAPABILITY_WORKFLOW_RUNTIME_WORKER", + "WorkflowCurrentStepResolution", "WorkflowDefinitionContribution", "WorkflowDefinitionContributionProvider", + "WorkflowInstanceRef", + "WorkflowOrchestrationProvider", "WorkflowRuntimeWorker", + "WorkflowStandardStartRequest", "workflow_definition_contribution_hash", "workflow_definition_contribution_provider", + "workflow_orchestration_provider", "workflow_runtime_worker", ] diff --git a/src/govoplan_core/server/default_config.py b/src/govoplan_core/server/default_config.py index ea21196..d89feda 100644 --- a/src/govoplan_core/server/default_config.py +++ b/src/govoplan_core/server/default_config.py @@ -35,6 +35,12 @@ async def lifespan(app: FastAPI): api_key_secret=settings.dev_bootstrap_api_key, user_password=settings.dev_bootstrap_password, ) + lifecycle = getattr(app.state, "govoplan_lifecycle", None) + if lifecycle is not None and hasattr( + lifecycle, + "reconcile_workflow_definitions", + ): + lifecycle.reconcile_workflow_definitions() yield diff --git a/tests/test_workflow_contribution_lifecycle.py b/tests/test_workflow_contribution_lifecycle.py new file mode 100644 index 0000000..c8c309f --- /dev/null +++ b/tests/test_workflow_contribution_lifecycle.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +import unittest + +from fastapi import FastAPI + +from govoplan_core.core.lifecycle import ModuleLifecycleManager +from govoplan_core.core.modules import ( + ModuleInterfaceProvider, + ModuleManifest, +) +from govoplan_core.core.registry import PlatformRegistry +from govoplan_core.core.workflows import ( + CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS, + WorkflowDefinitionContribution, +) +from govoplan_core.db.session import configure_database, reset_database + + +class _ContributionProvider: + def __init__(self) -> None: + self.calls = 0 + + def reconcile(self, session, *, tenant_ids=()): + del session, tenant_ids + self.calls += 1 + return {"created": 1, "updated": 0} + + +class WorkflowContributionLifecycleTests(unittest.TestCase): + def setUp(self) -> None: + configure_database("sqlite:///:memory:") + + def tearDown(self) -> None: + reset_database(dispose=True) + + def test_active_graph_change_reconciles_module_workflow_baselines(self) -> None: + provider = _ContributionProvider() + workflow_engine = ModuleManifest( + id="workflow_engine", + name="Workflow Engine", + version="1.0.0", + provides_interfaces=( + ModuleInterfaceProvider( + name="workflow.definition_contributions", + version="1.0.0", + ), + ), + capability_factories={ + CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS: ( + lambda _context: provider + ), + }, + ) + domain = ModuleManifest( + id="permits", + name="Permits", + version="1.0.0", + workflow_definitions=( + WorkflowDefinitionContribution( + origin_module_id="permits", + origin_module_version="1.0.0", + definition_key="approve", + name="Approve permit", + graph={"nodes": [], "edges": []}, + ), + ), + ) + registry = PlatformRegistry() + lifecycle = ModuleLifecycleManager( + registry=registry, + available_modules={ + workflow_engine.id: workflow_engine, + domain.id: domain, + }, + settings=None, + ) + lifecycle.attach_app(FastAPI()) + + result = lifecycle.apply_enabled_modules( + ("workflow_engine", "permits"), + migrate=False, + protected_modules=(), + ) + + self.assertEqual(("permits", "workflow_engine"), result.enabled_modules) + self.assertEqual(1, provider.calls) + + +if __name__ == "__main__": + unittest.main()