Add IDM assignment lifecycle worker contract

src/govoplan_core/celery_app.py

@@ -19,6 +19,10 @@ from govoplan_core.core.events import (
     PlatformEventOutbox,
     publish_platform_event,
 )
+from govoplan_core.core.idm import (
+    CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE,
+    IdmAssignmentLifecycle,
+)
 from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids
 from govoplan_core.core.mail import CAPABILITY_MAIL_DELIVERY_OUTBOX, MailDeliveryOutboxProvider
 from govoplan_core.core.modules import ModuleContext
@@ -62,6 +66,7 @@ celery.conf.update(
         "govoplan.postbox.dispatch_routes": {"queue": "postbox"},
         "govoplan.events.dispatch_outbox": {"queue": "events"},
         "govoplan.events.purge_outbox": {"queue": "events"},
+        "govoplan.idm.expire_assignments": {"queue": "idm"},
     },
     worker_prefetch_multiplier=1,
     task_acks_late=True,
@@ -117,6 +122,11 @@ celery.conf.update(
             "schedule": 24 * 60 * 60.0,
             "args": (500,),
         },
+        "idm-assignment-expiry-every-minute": {
+            "task": "govoplan.idm.expire_assignments",
+            "schedule": 60.0,
+            "args": (None, 100),
+        },
     },
 )
 
@@ -238,6 +248,20 @@ def _platform_event_outbox(
     return capability
 
 
+def _idm_assignment_lifecycle(
+    registry: PlatformRegistry | None = None,
+) -> IdmAssignmentLifecycle | None:
+    registry = registry or _platform_registry()
+    if not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE):
+        return None
+    capability = registry.require_capability(
+        CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE
+    )
+    if not isinstance(capability, IdmAssignmentLifecycle):
+        raise RuntimeError("IDM assignment lifecycle capability is invalid")
+    return capability
+
+
 @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0)
 def send_email(self, job_id: str):
     """Send one explicitly queued campaign job.
@@ -485,6 +509,35 @@ def dispatch_postbox_routes(
         return result
 
 
+@celery.task(
+    name="govoplan.idm.expire_assignments",
+    bind=True,
+    max_retries=0,
+)
+def expire_idm_assignments(
+    self,
+    tenant_id: str | None = None,
+    limit: int = 100,
+):
+    """Emit idempotent lifecycle events for elapsed IDM assignments."""
+
+    from govoplan_core.db.session import get_database
+
+    with get_database().SessionLocal() as session:
+        provider = _idm_assignment_lifecycle()
+        if provider is None:
+            return {"selected": 0, "expired": 0, "assignment_ids": []}
+        result = dict(
+            provider.process_expired(
+                session,
+                tenant_id=tenant_id,
+                limit=limit,
+            )
+        )
+        session.commit()
+        return result
+
+
 @celery.task(
     name="govoplan.events.dispatch_outbox",
     bind=True,

src/govoplan_core/core/idm.py

@@ -9,6 +9,7 @@ from typing import Literal, Protocol, runtime_checkable
 IDM_MODULE_ID = "idm"
 CAPABILITY_IDM_DIRECTORY = "idm.directory"
 CAPABILITY_IDM_FUNCTION_ASSIGNMENTS = "idm.function_assignments"
+CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE = "idm.assignmentLifecycle"
 
 IdmStatus = Literal["active", "inactive", "suspended"]
 OrganizationFunctionAssignmentSource = Literal["direct", "delegated", "acting_for", "directory", "governance", "system"]
@@ -106,3 +107,36 @@ class IdmFunctionAssignmentDirectory(Protocol):
         effective_at: datetime | None = None,
     ) -> Mapping[str, OrganizationFunctionIncumbencyRef]:
         ...
+
+
+@runtime_checkable
+class IdmAssignmentLifecycle(Protocol):
+    """Worker boundary for time-driven function-assignment transitions."""
+
+    def process_expired(
+        self,
+        session: object,
+        *,
+        tenant_id: str | None = None,
+        effective_at: datetime | None = None,
+        limit: int = 100,
+    ) -> Mapping[str, object]:
+        ...
+
+
+def idm_assignment_lifecycle(
+    registry: object | None,
+) -> IdmAssignmentLifecycle | None:
+    if (
+        registry is None
+        or not hasattr(registry, "has_capability")
+        or not hasattr(registry, "capability")
+        or not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE)
+    ):
+        return None
+    capability = registry.capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE)
+    return (
+        capability
+        if isinstance(capability, IdmAssignmentLifecycle)
+        else None
+    )

tests/test_idm_assignment_lifecycle_worker.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from govoplan_core.celery_app import celery, expire_idm_assignments
+from govoplan_core.core.idm import (
+    CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE,
+    IdmAssignmentLifecycle,
+    idm_assignment_lifecycle,
+)
+from govoplan_core.core.modules import ModuleContext, ModuleManifest
+from govoplan_core.core.registry import PlatformRegistry
+
+
+class _Lifecycle:
+    def process_expired(
+        self,
+        session,
+        *,
+        tenant_id=None,
+        effective_at=None,
+        limit=100,
+    ):
+        del session, tenant_id, effective_at, limit
+        return {"selected": 0, "expired": 0, "assignment_ids": []}
+
+
+class IdmAssignmentLifecycleWorkerTests(unittest.TestCase):
+    def test_contract_is_runtime_checkable_and_optional(self) -> None:
+        provider = _Lifecycle()
+        self.assertIsInstance(provider, IdmAssignmentLifecycle)
+        registry = PlatformRegistry()
+        registry.register(
+            ModuleManifest(
+                id="idm_lifecycle_test",
+                name="IDM lifecycle test",
+                version="test",
+                capability_factories={
+                    CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE: (
+                        lambda _context: provider
+                    )
+                },
+            )
+        )
+        registry.configure_capability_context(
+            ModuleContext(registry=registry, settings=object())
+        )
+
+        self.assertIs(provider, idm_assignment_lifecycle(registry))
+        self.assertIsNone(idm_assignment_lifecycle(PlatformRegistry()))
+
+    def test_worker_commits_provider_outcome(self) -> None:
+        session = MagicMock()
+        database = MagicMock()
+        database.SessionLocal.return_value.__enter__.return_value = session
+        provider = MagicMock()
+        provider.process_expired.return_value = {
+            "selected": 2,
+            "expired": 1,
+            "assignment_ids": ["assignment-1"],
+        }
+
+        with (
+            patch(
+                "govoplan_core.celery_app._idm_assignment_lifecycle",
+                return_value=provider,
+            ),
+            patch(
+                "govoplan_core.db.session.get_database",
+                return_value=database,
+            ),
+        ):
+            result = expire_idm_assignments.run("tenant-1", 25)
+
+        provider.process_expired.assert_called_once_with(
+            session,
+            tenant_id="tenant-1",
+            limit=25,
+        )
+        session.commit.assert_called_once_with()
+        self.assertEqual(1, result["expired"])
+
+    def test_route_and_periodic_job_are_registered(self) -> None:
+        self.assertEqual(
+            {"queue": "idm"},
+            celery.conf.task_routes["govoplan.idm.expire_assignments"],
+        )
+        schedule = celery.conf.beat_schedule[
+            "idm-assignment-expiry-every-minute"
+        ]
+        self.assertEqual("govoplan.idm.expire_assignments", schedule["task"])
+        self.assertEqual(60.0, schedule["schedule"])
+
+
+if __name__ == "__main__":
+    unittest.main()