Project durable workflow handoffs as work items

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")
 

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(

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"),

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")

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": ""},
     ),

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:

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"]