diff --git a/README.md b/README.md index 34b3620..158c1fb 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,13 @@ human handoffs; event filters and variable mappings are bounded JSON expressions and never executable code. Cron remains an optional governed scheduler-adapter concern. +External hand-off waits bind an immutable, revision-bearing domain reference +to a focused action URL and resume only from declared terminal platform events. +The runtime rechecks the provider's current authorization and resource revision +before continuing. Observational events are duplicate-safe, navigation alone +never completes work, and a missing optional projection remains visible without +changing the domain outcome. + Consequential module actions are staged in Core's durable recovery ledger before provider dispatch. Conclusive effects commit with the Workflow projection. Lost acknowledgements block continuation and expose evidence-based diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md index 5f60613..11fe33e 100644 --- a/docs/CONCEPT.md +++ b/docs/CONCEPT.md @@ -119,6 +119,9 @@ The first executable slice now provides: - activation-bound API, one-time/interval schedule, platform-event, and parent-workflow trigger registrations with exact-revision dispatch - durable duration/deadline/event wait subscriptions and scale-out-safe claims +- resumable external hand-offs with immutable references, focused action URLs, + duplicate-safe observed events, terminal outcome ports, timeout paths, and a + provider authorization/revision recheck before continuation - a separate transactional platform-event consumer with bounded JSON filters and variable mappings, idempotent delivery, and current-authority rechecks - Core recovery-ledger operations for module actions, including canonical diff --git a/pyproject.toml b/pyproject.toml index 708659f..8a04d27 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-workflow-engine" -version = "0.1.18" +version = "0.1.19" description = "Headless, versioned workflow definition and execution engine for GovOPlaN." readme = "README.md" requires-python = ">=3.12" license = "AGPL-3.0-or-later" authors = [{ name = "GovOPlaN" }] -dependencies = ["defusedxml>=0.7,<1", "govoplan-core>=0.1.18"] +dependencies = ["defusedxml>=0.7,<1", "govoplan-core>=0.1.28"] [tool.setuptools.packages.find] where = ["src"] diff --git a/src/govoplan_workflow_engine/backend/bpmn_graph.py b/src/govoplan_workflow_engine/backend/bpmn_graph.py index 0fa19d8..92087c3 100644 --- a/src/govoplan_workflow_engine/backend/bpmn_graph.py +++ b/src/govoplan_workflow_engine/backend/bpmn_graph.py @@ -228,6 +228,10 @@ def legacy_graph_to_bpmn(graph: WorkflowGraph) -> WorkflowGraph: config["wait_mode"] = config.pop("mode", "manual") config.setdefault("event_definition", "none") bpmn_type = "bpmn.intermediateCatchEvent" + elif node_type == "workflow.external_handoff": + config["wait_mode"] = "external_handoff" + config.setdefault("event_definition", "message") + bpmn_type = "bpmn.intermediateCatchEvent" elif node_type == "workflow.capability": config["implementation"] = "capability" bpmn_type = "bpmn.serviceTask" @@ -1002,8 +1006,13 @@ def _runtime_node(node: WorkflowNode) -> WorkflowNode: elif node.type in {"bpmn.task", "bpmn.userTask", "bpmn.manualTask"}: node_type = "workflow.activity" elif node.type in {"bpmn.receiveTask", "bpmn.intermediateCatchEvent"}: - node_type = "workflow.wait" - config["mode"] = config.get("wait_mode") or "event" + wait_mode = str(config.get("wait_mode") or "event") + node_type = ( + "workflow.external_handoff" + if wait_mode == "external_handoff" + else "workflow.wait" + ) + config["mode"] = wait_mode elif node.type in {"bpmn.serviceTask", "bpmn.sendTask"}: implementation = str(config.get("implementation") or "capability") node_type = ( diff --git a/src/govoplan_workflow_engine/backend/instance_service.py b/src/govoplan_workflow_engine/backend/instance_service.py index ebd55aa..c29048e 100644 --- a/src/govoplan_workflow_engine/backend/instance_service.py +++ b/src/govoplan_workflow_engine/backend/instance_service.py @@ -751,7 +751,7 @@ def cancel_instance( instance.id, exc, ) - if step.node_type == "workflow.wait": + if step.node_type in {"workflow.wait", "workflow.external_handoff"}: from govoplan_workflow_engine.backend.triggers import ( resolve_wait_state, ) @@ -1010,6 +1010,25 @@ def _drive_instance( registry=registry, ) return + if node.type == "workflow.external_handoff": + from govoplan_workflow_engine.backend.triggers import ( + register_external_handoff_state, + ) + + handoff_state = register_external_handoff_state( + session, + instance=instance, + step=step, + node=node, + ) + _set_external_handoff( + session, + instance=instance, + step=step, + node=node, + wait_state=handoff_state, + ) + return if node.type == "workflow.wait": from govoplan_workflow_engine.backend.triggers import ( register_wait_state, @@ -2340,6 +2359,51 @@ def _set_automated_wait( ) +def _set_external_handoff( + session: Session, + *, + instance: WorkflowInstance, + step: WorkflowInstanceStep, + node: WorkflowNode, + wait_state: object, +) -> None: + config = dict(getattr(wait_state, "config_")) + due_at = getattr(wait_state, "due_at") + optional_capabilities = { + str(key): bool(value) + for key, value in dict(config.get("optional_capabilities") or {}).items() + } + unavailable = sorted( + name for name, available in optional_capabilities.items() if not available + ) + step.status = "waiting" + step.external_ref = str(config.get("immutable_ref") or "") or None + step.handoff = { + "kind": "external_handoff", + "state": "assigned", + "title": str(node.config.get("title") or node.label or "External hand-off"), + "instructions": str(node.config.get("instructions") or ""), + "event_type": getattr(wait_state, "event_type"), + "external_id": config.get("external_id"), + "immutable_ref": config.get("immutable_ref"), + "action_url": config.get("action_url"), + "due_at": due_at.isoformat() if due_at else None, + "optional_capabilities": optional_capabilities, + "unavailable_optional_capabilities": unavailable, + "allowed_actions": [], + } + _clear_work_projection(step) + instance.status = "waiting" + _record_event( + session, + instance, + step=step, + kind="workflow.external_handoff.registered", + actor_id=instance.created_by, + payload=dict(step.handoff), + ) + + def _set_dependency_handoff( session: Session, *, @@ -2807,6 +2871,26 @@ def _require_runtime_dependencies( principal=principal, registry=registry, ) + if node.type == "workflow.external_handoff": + from govoplan_core.core.campaigns import ( + CampaignWorkOrchestrationProvider, + ) + + capability_name = str( + node.config.get("provider_capability") or "" + ).strip() + capability = ( + registry.capability(capability_name) + if capability_name + and registry is not None + and hasattr(registry, "has_capability") + and registry.has_capability(capability_name) + else None + ) + if not isinstance(capability, CampaignWorkOrchestrationProvider): + raise WorkflowConflictError( + f"External hand-off capability {capability_name!r} is not available." + ) def _authorization_payload( diff --git a/src/govoplan_workflow_engine/backend/manifest.py b/src/govoplan_workflow_engine/backend/manifest.py index 75a5486..95216bf 100644 --- a/src/govoplan_workflow_engine/backend/manifest.py +++ b/src/govoplan_workflow_engine/backend/manifest.py @@ -9,6 +9,7 @@ from govoplan_core.core.access import ( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_RUN_LIFECYCLE +from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY from govoplan_core.core.module_guards import ( drop_table_retirement_provider, @@ -16,6 +17,7 @@ from govoplan_core.core.module_guards import ( ) from govoplan_core.core.modules import ( CapabilityDocumentation, + DocumentationCondition, DocumentationTopic, MigrationSpec, ModuleContext, @@ -58,7 +60,7 @@ from govoplan_workflow_engine.backend.service_launcher import ( MODULE_ID = "workflow_engine" MODULE_NAME = "Workflow Engine" -MODULE_VERSION = "0.1.18" +MODULE_VERSION = "0.1.19" DEFINITION_READ_SCOPE = "workflow:definition:read" DEFINITION_WRITE_SCOPE = "workflow:definition:write" @@ -213,6 +215,7 @@ manifest = ModuleManifest( optional_dependencies=( "access", "audit", + "campaigns", "dataflow", "datasources", "idm", @@ -227,6 +230,7 @@ manifest = ModuleManifest( CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, + CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION, CAPABILITY_DATAFLOW_RUN_LIFECYCLE, CAPABILITY_IDM_DIRECTORY, CAPABILITY_NOTIFICATIONS_DISPATCH, @@ -269,6 +273,12 @@ manifest = ModuleManifest( ), ), requires_interfaces=( + ModuleInterfaceRequirement( + name="campaigns.work_orchestration", + version_min="1.0.0", + version_max_exclusive="2.0.0", + optional=True, + ), ModuleInterfaceRequirement( name=CAPABILITY_ACCESS_REFERENCE_OPTIONS, version_min="0.1.0", @@ -376,6 +386,50 @@ manifest = ModuleManifest( ), ), documentation=( + DocumentationTopic( + id="workflow.external-campaign-handoffs", + title="Resume Workflow from accountable Campaign work", + summary=( + "Open revision-bearing Campaign work, focus the relevant UI, " + "and continue from declared lifecycle events without polling." + ), + body=( + "The external hand-off primitive follows a module action that " + "creates or references a Campaign and opens one accountable work " + "assignment. Workflow stores the exact Campaign version and assignment " + "revision, a safe action link, correlation and idempotency provenance, " + "and a durable event subscription. Assigned, accepted, started, and " + "reassigned events update the waiting state; completed, rejected, " + "cancelled, and timed-out outcomes take separate graph paths. Duplicate " + "events are ignored. Before a terminal event resumes execution, Workflow " + "re-resolves its authority and Campaign rechecks both access and the event " + "revision. An assignment never grants Campaign access. Missing Campaign or " + "Views capabilities leave the instance inspectable and blocked; missing " + "Tasks or Notifications is reported as reduced optional integration and " + "does not change the Campaign outcome." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "workflow_designer", "campaign_manager", "module_admin"), + related_modules=("campaigns", "views", "tasks", "notifications", "audit"), + conditions=( + DocumentationCondition( + any_scopes=(INSTANCE_READ_SCOPE, ADMIN_SCOPE), + ), + ), + order=75, + metadata={ + "kind": "workflow", + "help_contexts": [ + "workflow.external-handoff", + "workflow.instances", + ], + "limitations": [ + "The Campaign provider must be installed to start or resume a Campaign hand-off.", + "Focused View projection is optional; the safe Campaign action link remains available without it.", + ], + }, + ), DocumentationTopic( id="workflow.data-subject-requests", title="Workflow data-subject requests", @@ -492,6 +546,7 @@ manifest = ModuleManifest( "workflow revision", "workflow instance", "work transition", + "external event hand-off", "execution adapter binding", ), non_owned_concepts=( diff --git a/src/govoplan_workflow_engine/backend/node_library.py b/src/govoplan_workflow_engine/backend/node_library.py index 3982086..7bdcf8e 100644 --- a/src/govoplan_workflow_engine/backend/node_library.py +++ b/src/govoplan_workflow_engine/backend/node_library.py @@ -269,6 +269,86 @@ LEGACY_WORKFLOW_NODE_TYPES = ( "view_surface_ids": [], }, ), + DefinitionNodeType( + type="workflow.external_handoff", + category="integration", + label="External hand-off", + description=( + "Wait for revision-bearing module events without polling, while " + "preserving a focused action link and explicit terminal outcomes." + ), + icon="arrow-left-right", + input_ports=(DefinitionPort(id="input", label="Input"),), + output_ports=( + DefinitionPort(id="completed", label="Completed", required=False), + DefinitionPort(id="rejected", label="Rejected", required=False), + DefinitionPort(id="cancelled", label="Cancelled", required=False), + DefinitionPort(id="timed_out", label="Timed out", required=False), + ), + config_fields=( + DefinitionConfigField( + id="provider_capability", + label="Hand-off provider", + kind="capability", + required=True, + ), + DefinitionConfigField( + id="event_type", + label="Event type", + kind="text", + required=True, + ), + DefinitionConfigField( + id="event_filter", + label="Event filter", + kind="mapping", + required=True, + ), + DefinitionConfigField( + id="external_id", + label="External reference path", + kind="expression", + required=True, + ), + DefinitionConfigField( + id="action_url", + label="Action URL path", + kind="expression", + required=True, + ), + DefinitionConfigField( + id="immutable_ref", + label="Immutable reference path", + kind="expression", + required=True, + ), + DefinitionConfigField( + id="timeout_after", + label="Timeout after", + kind="duration", + ), + FOCUSED_VIEW_SURFACES_FIELD, + ), + default_config={ + "provider_capability": "", + "event_type": "", + "event_filter": {}, + "outcome_path": "payload.outcome", + "terminal_outcomes": { + "completed": "completed", + "rejected": "rejected", + "cancelled": "cancelled", + }, + "observed_outcomes": ["assigned", "accepted", "started", "reassigned"], + "external_id": "", + "expected_revision": "", + "action_url": "", + "immutable_ref": "", + "optional_capabilities": {}, + "timeout_after": "", + "view_surface_ids": [], + }, + ), DefinitionNodeType( type="workflow.capability", category="integration", diff --git a/src/govoplan_workflow_engine/backend/triggers.py b/src/govoplan_workflow_engine/backend/triggers.py index b8e2fd4..cac5a7a 100644 --- a/src/govoplan_workflow_engine/backend/triggers.py +++ b/src/govoplan_workflow_engine/backend/triggers.py @@ -264,6 +264,151 @@ def register_wait_state( return state +def register_external_handoff_state( + session: Session, + *, + instance: WorkflowInstance, + step: WorkflowInstanceStep, + node: WorkflowNode, + now: datetime | None = None, +) -> WorkflowWaitState: + """Persist one event-driven external hand-off with a bounded timeout.""" + + current = _as_utc(now or utcnow()) + event_type = _validated_event_type(node.config.get("event_type")) + event_filter = _resolved_context_value( + _event_filter(node.config.get("event_filter")), + instance.context_, + ) + if not isinstance(event_filter, Mapping) or not event_filter: + raise WorkflowConflictError( + "External hand-offs require a non-empty event filter." + ) + terminal = node.config.get("terminal_outcomes") + if not isinstance(terminal, Mapping) or not terminal: + raise WorkflowConflictError( + "External hand-offs require terminal outcome mappings." + ) + terminal_outcomes = { + str(key).strip(): str(value).strip() + for key, value in terminal.items() + if str(key).strip() and str(value).strip() + } + allowed_ports = {"completed", "rejected", "cancelled"} + if ( + not terminal_outcomes + or any(port not in allowed_ports for port in terminal_outcomes.values()) + ): + raise WorkflowConflictError( + "External hand-off outcomes must map to completed, rejected, or cancelled." + ) + observed = tuple( + dict.fromkeys( + str(value).strip() + for value in node.config.get("observed_outcomes") or () + if str(value).strip() + ) + ) + overlap = set(observed) & set(terminal_outcomes) + if overlap: + raise WorkflowConflictError( + "External hand-off outcomes cannot be both observed and terminal: " + + ", ".join(sorted(overlap)) + ) + timeout_after = _resolved_context_value( + node.config.get("timeout_after"), + instance.context_, + ) + due_at = ( + current + timedelta(seconds=_duration_seconds(timeout_after, minimum=1)) + if str(timeout_after or "").strip() + else None + ) + provider_capability = str( + _resolved_context_value( + node.config.get("provider_capability"), + instance.context_, + ) + or "" + ).strip() + external_id = str( + _resolved_context_value( + node.config.get("external_id"), + instance.context_, + ) + or "" + ).strip() + if not provider_capability or not external_id: + raise WorkflowConflictError( + "External hand-offs require a provider capability and external reference." + ) + expected_revision_value = _resolved_context_value( + node.config.get("expected_revision"), + instance.context_, + ) + expected_revision = ( + int(expected_revision_value) + if expected_revision_value not in (None, "") + else None + ) + if expected_revision is not None and expected_revision < 1: + raise WorkflowConflictError( + "External hand-off revisions start at one." + ) + action_url = _safe_action_url( + _resolved_context_value( + node.config.get("action_url"), + instance.context_, + ) + ) + immutable_ref = str( + _resolved_context_value( + node.config.get("immutable_ref"), + instance.context_, + ) + or "" + ).strip()[:1_000] + if not immutable_ref: + raise WorkflowConflictError( + "External hand-offs require an immutable reference." + ) + optional_capabilities = _resolved_context_value( + node.config.get("optional_capabilities") or {}, + instance.context_, + ) + if not isinstance(optional_capabilities, Mapping): + raise WorkflowConflictError( + "External hand-off capability availability must be an object." + ) + state = WorkflowWaitState( + tenant_id=instance.tenant_id, + instance_id=instance.id, + step_id=step.id, + mode="external_handoff", + status="waiting", + due_at=due_at, + event_type=event_type, + config_={ + "filter": dict(event_filter), + "outcome_path": str( + node.config.get("outcome_path") or "payload.outcome" + ).strip(), + "terminal_outcomes": terminal_outcomes, + "observed_outcomes": list(observed), + "observed_event_ids": [], + "provider_capability": provider_capability, + "external_id": external_id, + "expected_revision": expected_revision, + "action_url": action_url, + "immutable_ref": immutable_ref, + "optional_capabilities": dict(optional_capabilities), + }, + ) + session.add(state) + session.flush() + return state + + def resolve_wait_state( session: Session, *, @@ -328,7 +473,7 @@ def ingest_platform_event( .where( WorkflowWaitState.tenant_id == tenant_id, WorkflowWaitState.status == "waiting", - WorkflowWaitState.mode == "event", + WorkflowWaitState.mode.in_(("event", "external_handoff")), WorkflowWaitState.event_type == event.type, ) .with_for_update(skip_locked=True) @@ -338,6 +483,71 @@ def ingest_platform_event( for state in waits: if not _matches_filter(envelope, state.config_.get("filter")): continue + if state.mode == "external_handoff": + observed_ids = [ + str(value) + for value in state.config_.get("observed_event_ids") or () + ] + if event.event_id in observed_ids: + continue + outcome = str( + _value_at_path( + envelope, + state.config_.get("outcome_path") or "payload.outcome", + ) + or "" + ).strip() + terminal = state.config_.get("terminal_outcomes") + terminal_port = ( + str(terminal.get(outcome) or "").strip() + if isinstance(terminal, Mapping) + else "" + ) + observed_outcomes = { + str(value).strip() + for value in state.config_.get("observed_outcomes") or () + if str(value).strip() + } + if not terminal_port and outcome not in observed_outcomes: + continue + state.config_ = { + **dict(state.config_), + "observed_event_ids": [*observed_ids[-99:], event.event_id], + "last_outcome": outcome, + **( + {"selected_port": terminal_port} + if terminal_port + else {} + ), + } + if not terminal_port: + step = session.get(WorkflowInstanceStep, state.step_id) + instance = session.get(WorkflowInstance, state.instance_id) + if step is not None and instance is not None: + step.handoff = { + **dict(step.handoff), + "state": outcome, + "last_event_id": event.event_id, + } + from govoplan_workflow_engine.backend.instance_service import ( + _record_event, + ) + + _record_event( + session, + instance, + step=step, + kind="workflow.external_handoff.observed", + actor_id=event.actor.id if event.actor else None, + payload={ + "event_id": event.event_id, + "event_type": event.type, + "outcome": outcome, + "external_id": state.config_.get("external_id"), + }, + ) + state.revision += 1 + continue state.status = "triggered" state.source_event_id = event.event_id state.event_ = envelope @@ -700,11 +910,45 @@ def _dispatch_waits( continue graph = _runtime_graph(revision) triggered = state.status == "triggered" + external_inspection: dict[str, object] | None = None + if triggered and state.mode == "external_handoff": + external_inspection = _inspect_external_handoff( + session, + state=state, + step=step, + principal=principal, + registry=registry, + ) + if external_inspection.get("allowed") is not True: + reason = str( + external_inspection.get("reason") + or "External hand-off authorization is unavailable." + ) + state.error = reason + step.handoff = { + **dict(step.handoff), + "state": "blocked", + "message": reason, + "inspection": external_inspection, + } + skipped += 1 + release_workflow_state_fence(session, fence) + continue state.status = "resumed" if triggered else "timed_out" state.error = None state.revision += 1 output = ( - {"event": dict(state.event_ or {})} + { + "event": dict(state.event_ or {}), + **( + { + "external_handoff": external_inspection, + "immutable_ref": state.config_.get("immutable_ref"), + } + if external_inspection is not None + else {} + ), + } if triggered else {"due_at": state.due_at.isoformat() if state.due_at else None} ) @@ -713,8 +957,12 @@ def _dispatch_waits( instance, step=step, kind=( - "workflow.wait.event_received" + "workflow.external_handoff.completed" + if triggered and state.mode == "external_handoff" + else "workflow.wait.event_received" if triggered + else "workflow.external_handoff.timed_out" + if state.mode == "external_handoff" else "workflow.wait.timed_out" ), actor_id=None, @@ -725,7 +973,13 @@ def _dispatch_waits( instance=instance, step=step, graph=graph, - port="resumed" if triggered else "timed_out", + port=( + str(state.config_.get("selected_port") or "completed") + if triggered and state.mode == "external_handoff" + else "resumed" + if triggered + else "timed_out" + ), output=output, actor_id=None, ) @@ -751,6 +1005,92 @@ def _dispatch_waits( } +def _inspect_external_handoff( + session: Session, + *, + state: WorkflowWaitState, + step: WorkflowInstanceStep, + principal: ApiPrincipal, + registry: object | None, +) -> dict[str, object]: + from govoplan_core.core.campaigns import CampaignWorkOrchestrationProvider + + capability_name = str( + state.config_.get("provider_capability") or "" + ).strip() + if ( + not capability_name + or registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(capability_name) + ): + return { + "allowed": False, + "reason": ( + f"External hand-off capability {capability_name!r} is unavailable." + ), + "code": "workflow_external_handoff_provider_unavailable", + } + provider = registry.capability(capability_name) + if not isinstance(provider, CampaignWorkOrchestrationProvider): + return { + "allowed": False, + "reason": "External hand-off provider has an incompatible contract.", + "code": "workflow_external_handoff_provider_invalid", + } + event_revision = _value_at_path( + state.event_ or {}, + "payload.assignment_revision", + ) + try: + expected_revision = ( + int(event_revision) + if event_revision not in (None, "") + else None + ) + except (TypeError, ValueError): + return { + "allowed": False, + "reason": "External hand-off event has no valid resource revision.", + "code": "workflow_external_handoff_event_revision_invalid", + } + inspection = provider.inspect_handoff( + session, + principal, + tenant_id=state.tenant_id, + assignment_id=str(state.config_.get("external_id") or ""), + expected_revision=expected_revision, + ) + payload: dict[str, object] = { + "allowed": inspection.allowed, + "status": inspection.status, + "assignment_revision": inspection.assignment_revision, + "action_url": inspection.action_url, + "assignment_ref": inspection.assignment_ref, + "reason": inspection.reason, + "provenance": dict(inspection.provenance), + } + selected_port = str(state.config_.get("selected_port") or "") + if inspection.allowed and inspection.status != selected_port: + payload.update( + { + "allowed": False, + "reason": ( + "External hand-off state does not match the terminal event; " + "reload and reconcile the provider." + ), + "code": "workflow_external_handoff_state_mismatch", + } + ) + if inspection.allowed and inspection.assignment_ref: + state.config_ = { + **dict(state.config_), + "immutable_ref": inspection.assignment_ref, + } + step.external_ref = inspection.assignment_ref + return payload + + def _resolve_trigger_principal( session: Session, *, @@ -978,6 +1318,70 @@ def _event_filter(value: object) -> dict[str, object]: return parsed +def _resolved_context_value( + value: object, + context: Mapping[str, object], + *, + depth: int = 0, +) -> object: + if depth > 10: + raise WorkflowConflictError( + "External hand-off configuration is nested too deeply." + ) + if isinstance(value, str) and value.startswith("$"): + path = value[1:].lstrip(".") + current: object = context + if not path: + return dict(context) + for segment in path.split("."): + if not isinstance(current, Mapping) or segment not in current: + raise WorkflowConflictError( + f"External hand-off input path {value!r} is unavailable." + ) + current = current[segment] + return current + if isinstance(value, Mapping): + return { + str(key): _resolved_context_value( + item, + context, + depth=depth + 1, + ) + for key, item in value.items() + } + if isinstance(value, list): + return [ + _resolved_context_value(item, context, depth=depth + 1) + for item in value + ] + return value + + +def _value_at_path(value: object, path: object) -> object | None: + current = value + for segment in str(path or "").strip().lstrip("$").lstrip(".").split("."): + if not segment: + continue + if not isinstance(current, Mapping) or segment not in current: + return None + current = current[segment] + return current + + +def _safe_action_url(value: object) -> str: + candidate = str(value or "").strip() + if ( + not candidate.startswith("/") + or candidate.startswith("//") + or "\\" in candidate + or any(ord(character) < 32 or ord(character) == 127 for character in candidate) + ): + raise WorkflowConflictError( + "External hand-off action URLs must be safe application-relative paths." + ) + return candidate[:1_500] + + def _validated_event_type(value: object) -> str: event_type = str(value or "").strip() if not _EVENT_TYPE.fullmatch(event_type): diff --git a/tests/test_triggers.py b/tests/test_triggers.py index 1633645..05b3441 100644 --- a/tests/test_triggers.py +++ b/tests/test_triggers.py @@ -12,6 +12,10 @@ from govoplan_core.core.access import ( PrincipalRef, ) from govoplan_core.core.automation import AutomationPrincipalResolution +from govoplan_core.core.campaigns import ( + CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION, + CampaignWorkHandoffInspection, +) from govoplan_core.core.events import EventTenantRef, PlatformEvent from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation from govoplan_core.core.runtime_coordination import ( @@ -39,6 +43,7 @@ from govoplan_workflow_engine.backend.schemas import ( WorkflowNode, ) from govoplan_workflow_engine.backend.service import ( + WorkflowConflictError, activate_definition, create_definition, ) @@ -82,17 +87,58 @@ class AutomationProvider: ) +class CampaignHandoffProvider: + def __init__(self) -> None: + self.allowed = True + self.status = "completed" + self.revision = 2 + self.inspections: list[tuple[str, int | None]] = [] + + def prepare_handoff(self, _session, _principal, *, request): + raise AssertionError("The external wait must not create Campaign work.") + + def inspect_handoff( + self, + _session, + _principal, + *, + tenant_id, + assignment_id, + expected_revision=None, + ): + assert tenant_id == "tenant-1" + self.inspections.append((assignment_id, expected_revision)) + return CampaignWorkHandoffInspection( + allowed=self.allowed, + status=self.status, + assignment_revision=self.revision, + action_url="/campaigns/campaign-1/work?assignment=assignment-1", + assignment_ref=f"campaign-work-assignment:assignment-1:r{self.revision}", + reason=None if self.allowed else "Campaign access was revoked.", + provenance={"access_rechecked": True}, + ) + + class Registry: def __init__(self) -> None: self.provider = AutomationProvider() + self.campaign = CampaignHandoffProvider() def has_capability(self, name: str) -> bool: - return name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + return ( + name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + or ( + name == CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION + and self.campaign is not None + ) + ) def capability(self, name: str): if not self.has_capability(name): raise KeyError(name) - return self.provider + if name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER: + return self.provider + return self.campaign def graph(start_type: str, *, wait: WorkflowNode | None = None) -> WorkflowGraph: @@ -123,6 +169,83 @@ def graph(start_type: str, *, wait: WorkflowNode | None = None) -> WorkflowGraph return WorkflowGraph(nodes=nodes, edges=edges) +def external_handoff_graph(*, timeout_after: str = "1h") -> WorkflowGraph: + return WorkflowGraph( + nodes=[ + WorkflowNode(id="start", type="workflow.start.manual"), + WorkflowNode( + id="campaign_work", + type="workflow.external_handoff", + label="Complete Campaign review", + config={ + "provider_capability": CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION, + "event_type": "campaign.work.changed", + "event_filter": { + "payload": {"assignment_id": "$input.assignment_id"} + }, + "outcome_path": "payload.outcome", + "terminal_outcomes": { + "completed": "completed", + "rejected": "rejected", + "cancelled": "cancelled", + }, + "observed_outcomes": ["assigned", "accepted", "reassigned"], + "external_id": "$input.assignment_id", + "expected_revision": "$input.assignment_revision", + "action_url": "$input.action_url", + "immutable_ref": "$input.assignment_ref", + "optional_capabilities": "$input.optional_capabilities", + "timeout_after": timeout_after, + "view_surface_ids": ["campaigns.page.work"], + }, + ), + WorkflowNode(id="completed", type="workflow.end.completed"), + WorkflowNode( + id="rejected", + type="workflow.end.cancelled", + config={"reason": "Campaign work rejected"}, + ), + WorkflowNode( + id="cancelled", + type="workflow.end.cancelled", + config={"reason": "Campaign work cancelled"}, + ), + WorkflowNode( + id="timed_out", + type="workflow.end.cancelled", + config={"reason": "Campaign work timed out"}, + ), + ], + edges=[ + WorkflowEdge(id="start-work", source="start", target="campaign_work"), + WorkflowEdge( + id="work-completed", + source="campaign_work", + source_port="completed", + target="completed", + ), + WorkflowEdge( + id="work-rejected", + source="campaign_work", + source_port="rejected", + target="rejected", + ), + WorkflowEdge( + id="work-cancelled", + source="campaign_work", + source_port="cancelled", + target="cancelled", + ), + WorkflowEdge( + id="work-timeout", + source="campaign_work", + source_port="timed_out", + target="timed_out", + ), + ], + ) + + class WorkflowTriggerTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") @@ -175,6 +298,18 @@ class WorkflowTriggerTests(unittest.TestCase): ) return definition + def _external_input(self) -> dict[str, object]: + return { + "assignment_id": "assignment-1", + "assignment_revision": 1, + "action_url": "/campaigns/campaign-1/work?assignment=assignment-1", + "assignment_ref": "campaign-work-assignment:assignment-1:r1", + "optional_capabilities": { + "tasks": False, + "notifications": False, + }, + } + def test_schedule_registration_dispatch_and_replay_are_durable(self) -> None: definition = self._definition( graph("workflow.start.schedule"), @@ -280,6 +415,184 @@ class WorkflowTriggerTests(unittest.TestCase): self.assertEqual(1, result["waits_timed_out"]) self.assertEqual("completed", instance.status) + def test_external_handoff_observes_duplicate_safe_events_and_resumes(self) -> None: + definition = self._definition( + external_handoff_graph(), + automation=False, + ) + instance, _replayed = start_instance( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key="campaign-handoff-1", + input=self._external_input(), + ), + ) + step = self.session.get(WorkflowInstanceStep, instance.current_step_id) + state = self.session.scalar(select(WorkflowWaitState)) + assert step is not None and state is not None + self.assertEqual("external_handoff", state.mode) + self.assertEqual("assigned", step.handoff["state"]) + self.assertEqual( + ["notifications", "tasks"], + step.handoff["unavailable_optional_capabilities"], + ) + dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry) + accepted = PlatformEvent( + type="campaign.work.changed", + module_id="campaigns", + event_id="campaign-event-accepted", + tenant=EventTenantRef(id="tenant-1"), + payload={ + "assignment_id": "assignment-1", + "assignment_revision": 2, + "outcome": "accepted", + }, + ) + + observed = dispatcher.ingest_event(self.session, event=accepted) + duplicate = dispatcher.ingest_event(self.session, event=accepted) + + self.assertEqual(0, observed["waits_triggered"]) + self.assertEqual(0, duplicate["waits_triggered"]) + self.assertEqual("accepted", step.handoff["state"]) + self.assertEqual( + 1, + self.session.query(WorkflowInstanceEvent) + .filter( + WorkflowInstanceEvent.kind + == "workflow.external_handoff.observed" + ) + .count(), + ) + + completed = dispatcher.ingest_event( + self.session, + event=PlatformEvent( + type="campaign.work.changed", + module_id="campaigns", + event_id="campaign-event-completed", + tenant=EventTenantRef(id="tenant-1"), + payload={ + "assignment_id": "assignment-1", + "assignment_revision": 2, + "outcome": "completed", + }, + ), + ) + result = dispatcher.dispatch_due(self.session) + + self.assertEqual(1, completed["waits_triggered"]) + self.assertEqual(1, result["waits_resumed"]) + self.assertEqual("completed", instance.status) + self.assertEqual( + [("assignment-1", 2)], + self.registry.campaign.inspections, + ) + self.assertEqual( + "campaign-work-assignment:assignment-1:r2", + step.external_ref, + ) + + def test_external_handoff_revoked_access_blocks_until_rechecked(self) -> None: + definition = self._definition( + external_handoff_graph(), + automation=False, + ) + instance, _replayed = start_instance( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key="campaign-handoff-revoked", + input=self._external_input(), + ), + ) + dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry) + dispatcher.ingest_event( + self.session, + event=PlatformEvent( + type="campaign.work.changed", + module_id="campaigns", + tenant=EventTenantRef(id="tenant-1"), + payload={ + "assignment_id": "assignment-1", + "assignment_revision": 2, + "outcome": "completed", + }, + ), + ) + self.registry.campaign.allowed = False + + blocked = dispatcher.dispatch_due(self.session) + step = self.session.get(WorkflowInstanceStep, instance.current_step_id) + assert step is not None + + self.assertEqual(1, blocked["waits_skipped"]) + self.assertEqual("waiting", instance.status) + self.assertEqual("blocked", step.handoff["state"]) + self.assertIn("revoked", str(step.handoff["message"])) + + self.registry.campaign.allowed = True + resumed = dispatcher.dispatch_due(self.session) + + self.assertEqual(1, resumed["waits_resumed"]) + self.assertEqual("completed", instance.status) + + def test_external_handoff_timeout_and_optional_provider_absence(self) -> None: + definition = self._definition( + external_handoff_graph(timeout_after="1s"), + automation=False, + ) + instance, _replayed = start_instance( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key="campaign-handoff-timeout", + input=self._external_input(), + ), + ) + + result = SqlWorkflowTriggerDispatcher( + registry=self.registry + ).dispatch_due( + self.session, + now=datetime.now(tz=UTC) + timedelta(seconds=2), + ) + + self.assertEqual(1, result["waits_timed_out"]) + self.assertEqual("cancelled", instance.status) + + unavailable = self._definition( + external_handoff_graph(), + automation=False, + ) + self.registry.campaign = None # type: ignore[assignment] + with self.assertRaisesRegex(WorkflowConflictError, "is not available"): + start_instance( + self.session, + tenant_id="tenant-1", + definition_id=unavailable.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowInstanceStartRequest( + idempotency_key="campaign-handoff-unavailable", + input=self._external_input(), + ), + ) + def test_parent_workflow_outcome_starts_pinned_child(self) -> None: parent = self._definition( graph("workflow.start.manual"),