diff --git a/docs/ENGINE_EDITOR_SPLIT.md b/docs/ENGINE_EDITOR_SPLIT.md index 6e5c1ff..be5da93 100644 --- a/docs/ENGINE_EDITOR_SPLIT.md +++ b/docs/ENGINE_EDITOR_SPLIT.md @@ -135,3 +135,21 @@ baseline revision inactive when an older revision is active, and fails closed when required capabilities or interfaces are absent. Baselines are immutable; editing derives a pinned local override, and reset archives that override without removing revision or instance history. + +## Common Work Inbox + +Workflow Engine also projects the current human handoff of a waiting instance +through Core's versioned work-item provider contract. Tasks may aggregate that +projection when it is enabled, while Workflow Engine remains the sole owner of +the instance, step, allowed actions, and completion state. + +Human activities and reviews accept account, group, role, function, and +function-assignment responsibility references. A missing assignee defaults to +the account that started the instance. `due_after` uses the same bounded +duration syntax as Workflow timers (`5m`, `2h`, `1d`, or ISO 8601). The Engine +materializes assignment and due-date columns on the current step so inbox reads +remain tenant-scoped and indexed; historical JSON handoffs are interpreted +conservatively for upgrade compatibility. Automated waits and in-flight +provider calls are not presented as human work. Recovery-required, unknown, +dependency, and failed handoffs appear as blocked work and always link back to +the pinned Workflow instance. diff --git a/src/govoplan_workflow_engine/backend/db/models.py b/src/govoplan_workflow_engine/backend/db/models.py index ce03b32..ac31c7a 100644 --- a/src/govoplan_workflow_engine/backend/db/models.py +++ b/src/govoplan_workflow_engine/backend/db/models.py @@ -408,6 +408,13 @@ class WorkflowInstanceStep(Base, TimestampMixin): "tenant_id", "status", ), + Index( + "ix_workflow_instance_steps_work_assignment", + "tenant_id", + "status", + "work_assignment_kind", + "work_assignment_id", + ), ) id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) @@ -465,6 +472,25 @@ class WorkflowInstanceStep(Base, TimestampMixin): String(255), nullable=True, ) + work_assignment_kind: Mapped[str | None] = mapped_column( + String(40), + nullable=True, + index=True, + ) + work_assignment_id: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + work_assignment_label: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + ) + work_due_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) instance: Mapped[WorkflowInstance] = relationship(back_populates="steps") diff --git a/src/govoplan_workflow_engine/backend/instance_service.py b/src/govoplan_workflow_engine/backend/instance_service.py index e219341..ebd55aa 100644 --- a/src/govoplan_workflow_engine/backend/instance_service.py +++ b/src/govoplan_workflow_engine/backend/instance_service.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Mapping -from datetime import datetime +from datetime import datetime, timedelta import hashlib import logging @@ -1294,8 +1294,7 @@ def _execute_capability_step( action_input=action_input, preview_payload=preview_payload, backup_reference=( - str(node.config.get("recovery_backup_reference") or "").strip() - or None + str(node.config.get("recovery_backup_reference") or "").strip() or None ), approval_reference=( str(node.config.get("recovery_approval_reference") or "").strip() @@ -1368,11 +1367,7 @@ def _execute_capability_step( "operation_id": exc.operation_id, "status": recovery_status, "requires_attention": outcome_unknown, - **( - {"next_call_number": call_number + 1} - if safe_to_retry - else {} - ), + **({"next_call_number": call_number + 1} if safe_to_retry else {}), }, }, ) @@ -1473,9 +1468,7 @@ def _execute_capability_step( action_recovery.commit_unknown( session, error_type=type(exc).__name__, - message=( - "Inspect the provider by stable idempotency key before any retry" - ), + message=("Inspect the provider by stable idempotency key before any retry"), ) return True if not isinstance(result, ActionExecutionResult): @@ -1608,9 +1601,7 @@ def _execute_capability_step( session, provider_state=result.state, result_sha256=canonical_sha256(result_payload), - observed_effects_sha256=canonical_sha256( - result_payload["observed_effects"] - ), + observed_effects_sha256=canonical_sha256(result_payload["observed_effects"]), ) _drive_instance( session, @@ -1915,6 +1906,10 @@ def _set_action_handoff( "suggested_port": "failure", **details_payload, } + if state in {"pending", "running"}: + _clear_work_projection(step) + else: + _apply_work_projection(instance, step) instance.status = "waiting" instance.error = step.error if previous.get("state") != state or previous.get("message") != message: @@ -2109,6 +2104,7 @@ def _handle_dataflow_success( "warnings": warnings, "output": output, } + _apply_work_projection(instance, step) instance.status = "waiting" _record_event( session, @@ -2165,6 +2161,7 @@ def _complete_step( step.finished_at = utcnow() step.completed_by = actor_id step.handoff = {} + _clear_work_projection(step) context = dict(instance.context_) step_values = dict(context.get("steps") or {}) step_values[step.node_id] = dict(output) @@ -2183,6 +2180,80 @@ def _complete_step( return _next_node_id(graph, step.node_id, port) +_WORK_ASSIGNMENT_KINDS = { + "account", + "group", + "role", + "function", + "function_assignment", + "anyone", +} + + +def _work_assignment( + instance: WorkflowInstance, + configured: object | None = None, +) -> dict[str, str | None] | None: + if isinstance(configured, Mapping): + kind = str(configured.get("kind") or "").strip() + assignment_id = str(configured.get("id") or "").strip() + label = str(configured.get("label") or "").strip() or None + if kind in _WORK_ASSIGNMENT_KINDS and assignment_id: + if kind == "anyone" and assignment_id != "*": + return None + return {"kind": kind, "id": assignment_id, "label": label} + return None + + value = str(configured or "").strip() + if value: + prefix, separator, remainder = value.partition(":") + if separator and prefix in _WORK_ASSIGNMENT_KINDS and remainder.strip(): + assignment_id = remainder.strip() + if prefix == "anyone" and assignment_id != "*": + return None + return {"kind": prefix, "id": assignment_id, "label": None} + return {"kind": "account", "id": value, "label": None} + + account_id = str(instance.authorization_.get("account_id") or "").strip() + if not account_id: + return None + return {"kind": "account", "id": account_id, "label": None} + + +def _work_due_at(configured: object | None) -> datetime | None: + if not str(configured or "").strip(): + return None + from govoplan_workflow_engine.backend.triggers import duration_seconds + + return utcnow() + timedelta(seconds=duration_seconds(configured)) + + +def _apply_work_projection( + instance: WorkflowInstance, + step: WorkflowInstanceStep, + *, + assignment: Mapping[str, object] | None = None, + due_at: datetime | None = None, +) -> None: + normalized = ( + dict(assignment) if assignment is not None else _work_assignment(instance) + ) + if normalized is None: + _clear_work_projection(step) + return + step.work_assignment_kind = str(normalized.get("kind") or "") or None + step.work_assignment_id = str(normalized.get("id") or "") or None + step.work_assignment_label = str(normalized.get("label") or "") or None + step.work_due_at = due_at + + +def _clear_work_projection(step: WorkflowInstanceStep) -> None: + step.work_assignment_kind = None + step.work_assignment_id = None + step.work_assignment_label = None + step.work_due_at = None + + def _set_human_handoff( session: Session, *, @@ -2200,6 +2271,11 @@ def _set_human_handoff( else: actions = ["complete", "cancel"] kind = "activity" + assignment = _work_assignment( + instance, + node.config.get("reviewer") or node.config.get("assignee"), + ) + due_at = _work_due_at(node.config.get("due_after")) step.status = "waiting" step.handoff = { "kind": kind, @@ -2207,9 +2283,12 @@ def _set_human_handoff( "title": str(node.config.get("title") or node.label or node.type), "instructions": str(node.config.get("instructions") or ""), "assignee": node.config.get("reviewer") or node.config.get("assignee"), + "assignment": assignment, + "due_at": due_at.isoformat() if due_at else None, "required_evidence": list(node.config.get("required_evidence") or []), "allowed_actions": actions, } + _apply_work_projection(instance, step, assignment=assignment, due_at=due_at) instance.status = "waiting" _record_event( session, @@ -2249,6 +2328,7 @@ def _set_automated_wait( "event_type": event_type, "allowed_actions": ["cancel"], } + _clear_work_projection(step) instance.status = "waiting" _record_event( session, @@ -2275,6 +2355,7 @@ def _set_dependency_handoff( "message": message, "allowed_actions": ["retry", "cancel"], } + _apply_work_projection(instance, step) instance.status = "waiting" instance.error = message _record_event( @@ -2382,6 +2463,7 @@ def _set_failure_handoff( "allowed_actions": ["retry", "reject", "cancel"], "suggested_port": "failure", } + _apply_work_projection(instance, step) instance.status = "waiting" instance.error = message _record_event( diff --git a/src/govoplan_workflow_engine/backend/manifest.py b/src/govoplan_workflow_engine/backend/manifest.py index 789885f..48bcb53 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.idm import CAPABILITY_IDM_DIRECTORY from govoplan_core.core.module_guards import ( drop_table_retirement_provider, persistent_table_uninstall_guard, @@ -33,6 +34,7 @@ from govoplan_core.core.notifications import ( ) from govoplan_core.core.references import CAPABILITY_ACCESS_REFERENCE_OPTIONS from govoplan_core.core.views import CAPABILITY_VIEWS_RESOLVER +from govoplan_core.core.tasks import WorkItemProviderRegistration from govoplan_core.core.workflows import ( CAPABILITY_WORKFLOW_DEFINITION_CONTRIBUTIONS, CAPABILITY_WORKFLOW_ORCHESTRATION, @@ -187,6 +189,12 @@ def _service_launcher(context: ModuleContext) -> WorkflowServiceLauncher: return WorkflowServiceLauncher(registry=context.registry) +def _work_items(context: ModuleContext): + from govoplan_workflow_engine.backend.work_items import WorkflowWorkItemProvider + + return WorkflowWorkItemProvider(registry=context.registry) + + manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, @@ -198,6 +206,7 @@ manifest = ModuleManifest( "audit", "dataflow", "datasources", + "idm", "notifications", "policy", "tasks", @@ -210,6 +219,7 @@ manifest = ModuleManifest( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_DATAFLOW_RUN_LIFECYCLE, + CAPABILITY_IDM_DIRECTORY, CAPABILITY_NOTIFICATIONS_DISPATCH, CAPABILITY_POLICY_DEFINITION_GOVERNANCE, CAPABILITY_VIEWS_RESOLVER, @@ -303,6 +313,13 @@ manifest = ModuleManifest( contract_version="0.1.0", ), }, + work_item_providers=( + WorkItemProviderRegistration( + id="workflow_engine.handoffs", + factory=_work_items, + order=20, + ), + ), migration_spec=MigrationSpec( module_id=MODULE_ID, metadata=Base.metadata, @@ -350,7 +367,9 @@ manifest = ModuleManifest( "implementation imports. Definitions are persisted as immutable graph " "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." + "revision from an exact Portal Service binding and records that provenance. " + "When Tasks is enabled, current human handoffs are projected into the common " + "work inbox with typed responsibility, due date, and a resumable source link." ), layer="available", documentation_types=("admin", "user"), diff --git a/src/govoplan_workflow_engine/backend/migrations/versions/8d5a2f7c1b4e_v0118_work_projections.py b/src/govoplan_workflow_engine/backend/migrations/versions/8d5a2f7c1b4e_v0118_work_projections.py new file mode 100644 index 0000000..e581ab6 --- /dev/null +++ b/src/govoplan_workflow_engine/backend/migrations/versions/8d5a2f7c1b4e_v0118_work_projections.py @@ -0,0 +1,65 @@ +"""v0.1.18 Workflow work projections. + +Revision ID: 8d5a2f7c1b4e +Revises: e4a1f8c2d7b6 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "8d5a2f7c1b4e" +down_revision = "e4a1f8c2d7b6" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("workflow_instance_steps") as batch_op: + batch_op.add_column( + sa.Column("work_assignment_kind", sa.String(length=40), nullable=True) + ) + batch_op.add_column( + sa.Column("work_assignment_id", sa.String(length=255), nullable=True) + ) + batch_op.add_column( + sa.Column("work_assignment_label", sa.String(length=500), nullable=True) + ) + batch_op.add_column( + sa.Column("work_due_at", sa.DateTime(timezone=True), nullable=True) + ) + batch_op.create_index( + "ix_workflow_instance_steps_work_assignment_kind", + ["work_assignment_kind"], + ) + batch_op.create_index( + "ix_workflow_instance_steps_work_assignment_id", + ["work_assignment_id"], + ) + batch_op.create_index( + "ix_workflow_instance_steps_work_due_at", + ["work_due_at"], + ) + batch_op.create_index( + "ix_workflow_instance_steps_work_assignment", + [ + "tenant_id", + "status", + "work_assignment_kind", + "work_assignment_id", + ], + ) + + +def downgrade() -> None: + with op.batch_alter_table("workflow_instance_steps") as batch_op: + batch_op.drop_index("ix_workflow_instance_steps_work_assignment") + batch_op.drop_index("ix_workflow_instance_steps_work_due_at") + batch_op.drop_index("ix_workflow_instance_steps_work_assignment_id") + batch_op.drop_index("ix_workflow_instance_steps_work_assignment_kind") + batch_op.drop_column("work_due_at") + batch_op.drop_column("work_assignment_label") + batch_op.drop_column("work_assignment_id") + batch_op.drop_column("work_assignment_kind") diff --git a/src/govoplan_workflow_engine/backend/node_library.py b/src/govoplan_workflow_engine/backend/node_library.py index 92ce1c3..3982086 100644 --- a/src/govoplan_workflow_engine/backend/node_library.py +++ b/src/govoplan_workflow_engine/backend/node_library.py @@ -84,8 +84,7 @@ LEGACY_WORKFLOW_NODE_TYPES = ( category="trigger", label="Parent workflow", description=( - "Start as a pinned child or dependency of another Workflow " - "instance." + "Start as a pinned child or dependency of another Workflow instance." ), icon="git-branch", default_config={ @@ -165,8 +164,12 @@ LEGACY_WORKFLOW_NODE_TYPES = ( icon="square-check-big", input_ports=(DefinitionPort(id="input", label="Input"),), config_fields=( - DefinitionConfigField(id="title", label="Title", kind="text", required=True), - DefinitionConfigField(id="instructions", label="Instructions", kind="textarea"), + DefinitionConfigField( + id="title", label="Title", kind="text", required=True + ), + DefinitionConfigField( + id="instructions", label="Instructions", kind="textarea" + ), DefinitionConfigField(id="assignee", label="Assignee", kind="subject"), DefinitionConfigField(id="due_after", label="Due after", kind="duration"), FOCUSED_VIEW_SURFACES_FIELD, @@ -192,8 +195,11 @@ LEGACY_WORKFLOW_NODE_TYPES = ( DefinitionPort(id="rejected", label="Rejected", required=False), ), config_fields=( - DefinitionConfigField(id="title", label="Title", kind="text", required=True), + DefinitionConfigField( + id="title", label="Title", kind="text", required=True + ), DefinitionConfigField(id="reviewer", label="Reviewer", kind="subject"), + DefinitionConfigField(id="due_after", label="Due after", kind="duration"), DefinitionConfigField( id="required_evidence", label="Required evidence", @@ -204,6 +210,7 @@ LEGACY_WORKFLOW_NODE_TYPES = ( default_config={ "title": "", "reviewer": "", + "due_after": "", "required_evidence": [], "view_surface_ids": [], }, @@ -292,7 +299,9 @@ LEGACY_WORKFLOW_NODE_TYPES = ( kind="text", required=True, ), - DefinitionConfigField(id="input_mapping", label="Input mapping", kind="mapping"), + DefinitionConfigField( + id="input_mapping", label="Input mapping", kind="mapping" + ), DefinitionConfigField( id="idempotency_key", label="Idempotency key", @@ -373,8 +382,7 @@ LEGACY_WORKFLOW_NODE_TYPES = ( label="Publication datasource", kind="text", description=( - "Optional stable Datasource target for materialized " - "output." + "Optional stable Datasource target for materialized output." ), ), DefinitionConfigField( @@ -397,7 +405,9 @@ LEGACY_WORKFLOW_NODE_TYPES = ( ("continue", "Follow failure path"), ), ), - DefinitionConfigField(id="input_mapping", label="Input mapping", kind="mapping"), + DefinitionConfigField( + id="input_mapping", label="Input mapping", kind="mapping" + ), FOCUSED_VIEW_SURFACES_FIELD, ), default_config={ @@ -428,7 +438,9 @@ LEGACY_WORKFLOW_NODE_TYPES = ( ), output_ports=(), config_fields=( - DefinitionConfigField(id="output_mapping", label="Output mapping", kind="mapping"), + DefinitionConfigField( + id="output_mapping", label="Output mapping", kind="mapping" + ), ), default_config={"output_mapping": {}}, ), @@ -870,7 +882,9 @@ BPMN_NODE_TYPES = ( shape="activity", config_fields=( *_TASK_FIELDS, - DefinitionConfigField(id="message_ref", label="Message reference", kind="text"), + DefinitionConfigField( + id="message_ref", label="Message reference", kind="text" + ), FOCUSED_VIEW_SURFACES_FIELD, ), default_config={ @@ -898,7 +912,13 @@ BPMN_NODE_TYPES = ( "Script task", "A BPMN script task retained as notation; arbitrary scripts are not executed.", "file-code-2", - (*_TASK_FIELDS, DefinitionConfigField(id="script_format", label="Script format", kind="text"), DefinitionConfigField(id="script", label="Script", kind="textarea")), + ( + *_TASK_FIELDS, + DefinitionConfigField( + id="script_format", label="Script format", kind="text" + ), + DefinitionConfigField(id="script", label="Script", kind="textarea"), + ), {"title": "", "instructions": "", "script_format": "", "script": ""}, ), ( @@ -906,7 +926,14 @@ BPMN_NODE_TYPES = ( "Business rule task", "Evaluate a governed business-rule implementation.", "scale", - (*_TASK_FIELDS, DefinitionConfigField(id="implementation_ref", label="Implementation reference", kind="text")), + ( + *_TASK_FIELDS, + DefinitionConfigField( + id="implementation_ref", + label="Implementation reference", + kind="text", + ), + ), {"title": "", "instructions": "", "implementation_ref": ""}, ), ( @@ -914,7 +941,15 @@ BPMN_NODE_TYPES = ( "Call activity", "Call another reusable BPMN process or GovOPlaN workflow.", "external-link", - (*_TASK_FIELDS, DefinitionConfigField(id="called_element", label="Called element", kind="text", required=True)), + ( + *_TASK_FIELDS, + DefinitionConfigField( + id="called_element", + label="Called element", + kind="text", + required=True, + ), + ), {"title": "", "instructions": "", "called_element": ""}, ), ( @@ -956,11 +991,41 @@ BPMN_NODE_TYPES = ( runtime_support=runtime_support, ) for type_name, label, description, icon, runtime_support in ( - ("exclusiveGateway", "Exclusive gateway", "Choose exactly one matching sequence flow.", "diamond", "native"), - ("parallelGateway", "Parallel gateway", "Split or join concurrent sequence flows.", "plus", "model_only"), - ("inclusiveGateway", "Inclusive gateway", "Choose one or more matching sequence flows.", "circle-plus", "model_only"), - ("eventBasedGateway", "Event-based gateway", "Choose a path according to the first caught event.", "radio-tower", "model_only"), - ("complexGateway", "Complex gateway", "Apply a complex activation condition.", "asterisk", "model_only"), + ( + "exclusiveGateway", + "Exclusive gateway", + "Choose exactly one matching sequence flow.", + "diamond", + "native", + ), + ( + "parallelGateway", + "Parallel gateway", + "Split or join concurrent sequence flows.", + "plus", + "model_only", + ), + ( + "inclusiveGateway", + "Inclusive gateway", + "Choose one or more matching sequence flows.", + "circle-plus", + "model_only", + ), + ( + "eventBasedGateway", + "Event-based gateway", + "Choose a path according to the first caught event.", + "radio-tower", + "model_only", + ), + ( + "complexGateway", + "Complex gateway", + "Apply a complex activation condition.", + "asterisk", + "model_only", + ), ) ), _bpmn_node( @@ -972,8 +1037,12 @@ BPMN_NODE_TYPES = ( shape="data-object", input_ports=_OPTIONAL_INCOMING, config_fields=( - DefinitionConfigField(id="data_object_ref", label="Data object reference", kind="text"), - DefinitionConfigField(id="item_subject_ref", label="Item definition", kind="text"), + DefinitionConfigField( + id="data_object_ref", label="Data object reference", kind="text" + ), + DefinitionConfigField( + id="item_subject_ref", label="Item definition", kind="text" + ), ), default_config={"data_object_ref": "", "item_subject_ref": ""}, ), @@ -986,8 +1055,12 @@ BPMN_NODE_TYPES = ( shape="data-store", input_ports=_OPTIONAL_INCOMING, config_fields=( - DefinitionConfigField(id="data_store_ref", label="Data store reference", kind="text"), - DefinitionConfigField(id="item_subject_ref", label="Item definition", kind="text"), + DefinitionConfigField( + id="data_store_ref", label="Data store reference", kind="text" + ), + DefinitionConfigField( + id="item_subject_ref", label="Item definition", kind="text" + ), ), default_config={"data_store_ref": "", "item_subject_ref": ""}, ), @@ -1000,7 +1073,9 @@ BPMN_NODE_TYPES = ( shape="participant", input_ports=_OPTIONAL_INCOMING, config_fields=( - DefinitionConfigField(id="process_ref", label="Process reference", kind="text"), + DefinitionConfigField( + id="process_ref", label="Process reference", kind="text" + ), ), default_config={"process_ref": ""}, ), @@ -1013,7 +1088,9 @@ BPMN_NODE_TYPES = ( shape="lane", input_ports=_OPTIONAL_INCOMING, config_fields=( - DefinitionConfigField(id="flow_node_refs", label="Flow node references", kind="string_list"), + DefinitionConfigField( + id="flow_node_refs", label="Flow node references", kind="string_list" + ), ), default_config={"flow_node_refs": []}, ), @@ -1040,7 +1117,9 @@ BPMN_NODE_TYPES = ( shape="group", input_ports=_OPTIONAL_INCOMING, config_fields=( - DefinitionConfigField(id="category_value_ref", label="Category value", kind="text"), + DefinitionConfigField( + id="category_value_ref", label="Category value", kind="text" + ), ), default_config={"category_value_ref": ""}, ), diff --git a/src/govoplan_workflow_engine/backend/triggers.py b/src/govoplan_workflow_engine/backend/triggers.py index e69ce2f..b8e2fd4 100644 --- a/src/govoplan_workflow_engine/backend/triggers.py +++ b/src/govoplan_workflow_engine/backend/triggers.py @@ -938,6 +938,12 @@ def _duration_seconds(value: object, *, minimum: int) -> int: return seconds +def duration_seconds(value: object, *, minimum: int = 1) -> int: + """Parse the duration syntax shared by timers and human-work due dates.""" + + return _duration_seconds(value, minimum=minimum) + + def _parse_instant(value: object, *, timezone_name: str) -> datetime: text = str(value or "").strip() if not text: diff --git a/src/govoplan_workflow_engine/backend/work_items.py b/src/govoplan_workflow_engine/backend/work_items.py new file mode 100644 index 0000000..e20c31f --- /dev/null +++ b/src/govoplan_workflow_engine/backend/work_items.py @@ -0,0 +1,367 @@ +from __future__ import annotations + +from collections.abc import Mapping +from datetime import UTC, datetime + +from sqlalchemy import and_, or_, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal, has_scope +from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, IdmDirectory +from govoplan_core.core.tasks import ( + WorkAssignmentRef, + WorkItem, + WorkItemPage, + WorkItemQuery, + WorkSourceRef, +) +from govoplan_workflow_engine.backend.db.models import ( + WorkflowDefinition, + WorkflowInstance, + WorkflowInstanceStep, +) +from govoplan_workflow_engine.backend.governance import definition_decision + + +PROVIDER_ID = "workflow_engine.handoffs" +INSTANCE_READ_SCOPE = "workflow:instance:read" +ADMIN_SCOPE = "workflow:instance:admin" +_NON_HUMAN_KINDS = {"timer", "event_wait", "dataflow_run"} +_BLOCKED_STATES = { + "blocked", + "failed", + "outcome_unknown", + "recovery_required", + "compensation_required", +} +_ASSIGNMENT_KINDS = { + "account", + "group", + "role", + "function", + "function_assignment", + "anyone", +} + + +class WorkflowWorkItemProvider: + def __init__(self, *, registry: object | None = None) -> None: + self.registry = registry + + def list_items( + self, + session: object, + principal: object, + *, + query: WorkItemQuery, + ) -> WorkItemPage: + if not isinstance(session, Session): + raise TypeError("Workflow work aggregation requires a SQLAlchemy Session.") + if not isinstance(principal, ApiPrincipal): + return WorkItemPage(items=(), total=0) + if principal.tenant_id != query.tenant_id: + return WorkItemPage(items=(), total=0) + administrative = has_scope(principal, ADMIN_SCOPE) + if not administrative and not has_scope(principal, INSTANCE_READ_SCOPE): + return WorkItemPage(items=(), total=0) + + statement = ( + select(WorkflowInstance, WorkflowInstanceStep, WorkflowDefinition) + .join( + WorkflowInstanceStep, + WorkflowInstanceStep.id == WorkflowInstance.current_step_id, + ) + .join( + WorkflowDefinition, + WorkflowDefinition.id == WorkflowInstance.definition_id, + ) + .where( + WorkflowInstance.tenant_id == query.tenant_id, + WorkflowInstance.status == "waiting", + WorkflowInstanceStep.status == "waiting", + ) + .order_by( + WorkflowInstanceStep.work_due_at.is_(None), + WorkflowInstanceStep.work_due_at.asc(), + WorkflowInstanceStep.updated_at.desc(), + WorkflowInstanceStep.id.desc(), + ) + ) + if not administrative: + targets = self._targets(principal, query.tenant_id) + conditions = [ + and_( + WorkflowInstanceStep.work_assignment_kind == kind, + WorkflowInstanceStep.work_assignment_id.in_(tuple(values)), + ) + for kind, values in targets.items() + if values + ] + conditions.append( + and_( + WorkflowInstanceStep.work_assignment_kind.is_(None), + WorkflowInstanceStep.work_assignment_id.is_(None), + ) + ) + statement = statement.where(or_(*conditions)) + + items: list[WorkItem] = [] + total = 0 + decisions: dict[str, bool] = {} + targets = self._targets(principal, query.tenant_id) + for instance, step, definition in session.execute(statement).yield_per(250): + assignment = _step_assignment(instance, step) + if not administrative and not _assignment_matches(assignment, targets): + continue + if not _is_actionable_handoff(step.handoff): + continue + allowed = decisions.get(definition.id) + if allowed is None: + allowed = definition_decision( + definition, + principal=principal, + registry=self.registry, + action="view", + ).allowed + decisions[definition.id] = allowed + if not allowed: + continue + item = _work_item(instance, step, definition, assignment) + if query.statuses and item.status not in query.statuses: + continue + if query.priorities and item.priority not in query.priorities: + continue + if query.due_before is not None and ( + item.due_at is None or _aware(item.due_at) > _aware(query.due_before) + ): + continue + if query.text and query.text.casefold() not in _search_text(item): + continue + total += 1 + if len(items) < query.limit: + items.append(item) + return WorkItemPage( + items=tuple(items), + total=total, + truncated=total > len(items), + ) + + def _targets(self, principal: ApiPrincipal, tenant_id: str) -> dict[str, set[str]]: + result = { + "account": {principal.account_id} if principal.account_id else set(), + "group": set(principal.group_ids), + "role": set(principal.role_ids), + "function_assignment": set(principal.function_assignment_ids), + "function": set(), + "anyone": {"*"}, + } + directory = self._idm_directory() + if directory is not None and principal.account_id: + result["function"].update( + item.function_id + for item in directory.organization_function_assignments_for_account( + principal.account_id, + tenant_id=tenant_id, + ) + if item.tenant_id == tenant_id and item.status == "active" + ) + return result + + def _idm_directory(self) -> IdmDirectory | None: + registry = self.registry + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_IDM_DIRECTORY) + ): + return None + provider = registry.capability(CAPABILITY_IDM_DIRECTORY) + return provider if isinstance(provider, IdmDirectory) else None + + +def _step_assignment( + instance: WorkflowInstance, + step: WorkflowInstanceStep, +) -> WorkAssignmentRef | None: + if step.work_assignment_kind and step.work_assignment_id: + return _assignment_ref( + step.work_assignment_kind, + step.work_assignment_id, + step.work_assignment_label, + ) + handoff_assignment = step.handoff.get("assignment") + if isinstance(handoff_assignment, Mapping): + kind = str(handoff_assignment.get("kind") or "").strip() + assignment_id = str(handoff_assignment.get("id") or "").strip() + if kind and assignment_id: + return _assignment_ref( + kind, + assignment_id, + str(handoff_assignment.get("label") or "").strip() or None, + ) + account_id = str(instance.authorization_.get("account_id") or "").strip() + return WorkAssignmentRef(kind="account", id=account_id) if account_id else None + + +def _assignment_matches( + assignment: WorkAssignmentRef | None, + targets: Mapping[str, set[str]], +) -> bool: + return bool( + assignment is not None and assignment.id in targets.get(assignment.kind, set()) + ) + + +def _is_actionable_handoff(handoff: Mapping[str, object]) -> bool: + kind = str(handoff.get("kind") or "") + state = str(handoff.get("state") or "waiting") + return kind not in _NON_HUMAN_KINDS and state not in {"pending", "running"} + + +def _work_item( + instance: WorkflowInstance, + step: WorkflowInstanceStep, + definition: WorkflowDefinition, + assignment: WorkAssignmentRef | None, +) -> WorkItem: + handoff = dict(step.handoff or {}) + state = str(handoff.get("state") or "waiting") + status = "blocked" if state in _BLOCKED_STATES else "open" + title = str( + handoff.get("title") or handoff.get("message") or f"Continue {definition.name}" + ) + required_action = str( + handoff.get("instructions") + or handoff.get("message") + or "Continue the current workflow handoff." + ).strip() + action_url = _action_url( + handoff.get("action_url") + or f"/workflow?definition={definition.id}&run={instance.id}" + ) + due_at = step.work_due_at or _date(handoff.get("due_at")) + priority = str(handoff.get("priority") or "normal").casefold() + if priority not in {"low", "normal", "high", "urgent"}: + priority = "normal" + updated_at = step.updated_at or instance.updated_at + revision = f"{step.attempt}:{updated_at.isoformat() if updated_at else '1'}" + return WorkItem( + id=step.id, + provider_id=PROVIDER_ID, + owner_module="workflow_engine", + tenant_id=instance.tenant_id, + title=title, + summary=f"{definition.name} ยท {step.node_type}", + status=status, # type: ignore[arg-type] + priority=priority, # type: ignore[arg-type] + required_action=required_action or None, + action_url=action_url, + due_at=due_at, + assignments=(assignment,) if assignment else (), + sources=( + WorkSourceRef( + module_id="workflow_engine", + resource_type="workflow_instance", + resource_id=instance.id, + revision=instance.definition_revision_id, + url=f"/workflow?definition={definition.id}&run={instance.id}", + label=definition.name, + ), + WorkSourceRef( + module_id="workflow_engine", + resource_type="workflow_step", + resource_id=step.id, + revision=str(step.attempt), + ), + ), + provenance={ + "definition_id": definition.id, + "definition_revision_id": instance.definition_revision_id, + "workflow_instance_id": instance.id, + "workflow_step_id": step.id, + }, + metadata={ + "handoff_kind": handoff.get("kind"), + "handoff_state": state, + "allowed_actions": _allowed_actions(handoff.get("allowed_actions")), + }, + revision=revision, + created_at=step.created_at, + updated_at=updated_at, + ) + + +def _assignment_ref( + kind: object, + assignment_id: object, + label: object = None, +) -> WorkAssignmentRef | None: + normalized_kind = str(kind or "").strip() + normalized_id = str(assignment_id or "").strip() + if normalized_kind not in _ASSIGNMENT_KINDS or not normalized_id: + return None + if normalized_kind == "anyone": + normalized_id = "*" + try: + normalized_label = str(label).strip()[:500] if label is not None else "" + return WorkAssignmentRef( + kind=normalized_kind, # type: ignore[arg-type] + id=normalized_id, + label=normalized_label or None, + ) + except ValueError: + return None + + +def _action_url(value: object) -> str: + candidate = str(value or "").strip() + if ( + candidate.startswith("/") + and not candidate.startswith("//") + and "\\" not in candidate + and all( + ord(character) >= 32 and ord(character) != 127 for character in candidate + ) + ): + return candidate[:1_500] + return "/workflow" + + +def _allowed_actions(value: object) -> list[str]: + if not isinstance(value, (list, tuple, set, frozenset)): + return [] + return [normalized for item in value if (normalized := str(item or "").strip())][ + :100 + ] + + +def _date(value: object) -> datetime | None: + if isinstance(value, datetime): + return value + text = str(value or "").strip() + if not text: + return None + try: + return datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + return None + + +def _aware(value: datetime) -> datetime: + return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) + + +def _search_text(item: WorkItem) -> str: + return " ".join( + value + for value in ( + item.title, + item.summary, + item.required_action, + item.owner_module, + ) + if value + ).casefold() + + +__all__ = ["PROVIDER_ID", "WorkflowWorkItemProvider"] diff --git a/tests/test_instance_service.py b/tests/test_instance_service.py index eb06900..c6d1e91 100644 --- a/tests/test_instance_service.py +++ b/tests/test_instance_service.py @@ -43,6 +43,7 @@ from govoplan_core.core.runtime_coordination import ( RuntimeIdentity, bind_process_runtime_identity, ) +from govoplan_core.core.tasks import WorkItemQuery from govoplan_core.db.base import Base from govoplan_core.db.base import utcnow from govoplan_workflow_engine.backend.db.models import ( @@ -84,6 +85,7 @@ from govoplan_workflow_engine.backend.service import ( create_definition, ) from govoplan_workflow_engine.backend.service_launcher import WorkflowServiceLauncher +from govoplan_workflow_engine.backend.work_items import WorkflowWorkItemProvider try: from test_bpmn import NATIVE_BPMN @@ -687,6 +689,107 @@ class WorkflowInstanceServiceTests(unittest.TestCase): start_origin="api", ) + def test_human_handoff_projects_typed_due_work_and_disappears_on_completion( + self, + ) -> None: + definition = create_definition( + self.session, + tenant_id="tenant-1", + actor_id="account-1", + payload=WorkflowDefinitionCreateRequest( + name="Guided case review", + graph=WorkflowGraph( + nodes=[ + WorkflowNode( + id="start", + type="workflow.start.manual", + config={"input_schema_ref": ""}, + ), + WorkflowNode( + id="activity", + type="workflow.activity", + config={ + "title": "Assess the application", + "instructions": "Record the assessment evidence.", + "assignee": "account:account-1", + "due_after": "2h", + }, + ), + WorkflowNode( + id="done", + type="workflow.end.completed", + ), + ], + edges=[ + WorkflowEdge( + id="start-activity", source="start", target="activity" + ), + WorkflowEdge( + id="activity-done", source="activity", target="done" + ), + ], + ), + execution_mode="guided", + ), + ) + activate_definition( + self.session, + tenant_id="tenant-1", + definition_id=definition.id, + actor_id="account-1", + ) + before = utcnow() + 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="guided-work-1"), + ) + step = self.session.get(WorkflowInstanceStep, instance.current_step_id) + self.assertIsNotNone(step) + assert step is not None + self.assertEqual("account", step.work_assignment_kind) + self.assertEqual("account-1", step.work_assignment_id) + self.assertIsNotNone(step.work_due_at) + assert step.work_due_at is not None + due_at = ( + step.work_due_at.replace(tzinfo=UTC) + if step.work_due_at.tzinfo is None + else step.work_due_at + ) + self.assertGreaterEqual(due_at, before + timedelta(hours=1, minutes=59)) + + provider = WorkflowWorkItemProvider(registry=self.registry) + page = provider.list_items( + self.session, + principal(), + query=WorkItemQuery(tenant_id="tenant-1"), + ) + self.assertEqual(1, page.total) + self.assertEqual("Assess the application", page.items[0].title) + self.assertEqual("account-1", page.items[0].assignments[0].id) + self.assertEqual(step.id, page.items[0].id) + + resolve_step( + self.session, + tenant_id="tenant-1", + instance_id=instance.id, + step_id=step.id, + actor_id="account-1", + principal=principal(), + registry=self.registry, + payload=WorkflowStepActionRequest(action="complete"), + ) + closed_page = provider.list_items( + self.session, + principal(), + query=WorkItemQuery(tenant_id="tenant-1"), + ) + self.assertEqual(0, closed_page.total) + def test_module_action_records_effects_and_completes_idempotently( self, ) -> None: @@ -990,9 +1093,7 @@ class WorkflowInstanceServiceTests(unittest.TestCase): request=request, ) checkpoint = session.scalar( - select(RecoveryCheckpoint).order_by( - RecoveryCheckpoint.sequence - ) + select(RecoveryCheckpoint).order_by(RecoveryCheckpoint.sequence) ) assert checkpoint is not None checkpoint.summary = "tampered provider evidence" diff --git a/tests/test_migrations.py b/tests/test_migrations.py index f0a56a9..fa9c289 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -29,7 +29,7 @@ class WorkflowMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "e4a1f8c2d7b6", + "8d5a2f7c1b4e", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertEqual( @@ -102,7 +102,12 @@ class WorkflowMigrationTests(unittest.TestCase): ) for path in current_revisions.glob("*.py"): if path.name.startswith( - ("0b4e7c9a2d6f_", "b2e4f6a8c0d1_", "e4a1f8c2d7b6_") + ( + "0b4e7c9a2d6f_", + "b2e4f6a8c0d1_", + "e4a1f8c2d7b6_", + "8d5a2f7c1b4e_", + ) ): continue shutil.copy2(path, legacy_revisions / path.name) @@ -146,7 +151,7 @@ class WorkflowMigrationTests(unittest.TestCase): manifest_factories=(get_manifest,), ) - self.assertIn("e4a1f8c2d7b6", result.current_revision or "") + self.assertIn("8d5a2f7c1b4e", result.current_revision or "") engine = create_engine(url) try: upgraded_tables = set(inspect(engine).get_table_names()) @@ -181,6 +186,25 @@ class WorkflowMigrationTests(unittest.TestCase): engine = create_engine(url) try: with engine.begin() as connection: + for index_name in ( + "ix_workflow_instance_steps_work_assignment", + "ix_workflow_instance_steps_work_due_at", + "ix_workflow_instance_steps_work_assignment_id", + "ix_workflow_instance_steps_work_assignment_kind", + ): + connection.execute(text(f"DROP INDEX {index_name}")) + for column_name in ( + "work_due_at", + "work_assignment_label", + "work_assignment_id", + "work_assignment_kind", + ): + connection.execute( + text( + "ALTER TABLE workflow_instance_steps " + f"DROP COLUMN {column_name}" + ) + ) connection.execute( text( "ALTER TABLE workflow_definition_revisions " @@ -196,7 +220,7 @@ class WorkflowMigrationTests(unittest.TestCase): connection.execute( text( "UPDATE alembic_version SET version_num = " - "'b2e4f6a8c0d1' WHERE version_num = 'e4a1f8c2d7b6'" + "'b2e4f6a8c0d1' WHERE version_num = '8d5a2f7c1b4e'" ) ) finally: @@ -218,9 +242,23 @@ class WorkflowMigrationTests(unittest.TestCase): } self.assertIn("bpmn_runtime_kind", columns) self.assertIn("bpmn_executable", columns) + step_columns = { + item["name"] + for item in inspect(engine).get_columns( + "workflow_instance_steps" + ) + } + self.assertTrue( + { + "work_assignment_kind", + "work_assignment_id", + "work_assignment_label", + "work_due_at", + }.issubset(step_columns) + ) with engine.connect() as connection: self.assertIn( - "e4a1f8c2d7b6", + "8d5a2f7c1b4e", set(MigrationContext.configure(connection).get_current_heads()), ) finally: