feat(core): dispatch durable dataflow runs

src/govoplan_core/celery_app.py

@@ -7,7 +7,9 @@ from celery import Celery
 from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CampaignDeliveryTaskProvider
 from govoplan_core.core.calendar import CAPABILITY_CALENDAR_OUTBOX, CalendarOutboxProvider
 from govoplan_core.core.dataflows import (
+    CAPABILITY_DATAFLOW_RUN_WORKER,
     CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER,
+    DataflowRunWorker,
     DataflowTriggerDispatcher,
 )
 from govoplan_core.core.events import (
@@ -42,6 +44,8 @@ celery.conf.update(
         "govoplan.notifications.deliver": {"queue": "notifications"},
         "govoplan.notifications.deliver_pending": {"queue": "notifications"},
         "govoplan.calendar.dispatch_outbox": {"queue": "calendar"},
+        "govoplan.dataflow.dispatch_runs": {"queue": "dataflow"},
+        "govoplan.dataflow.purge_runs": {"queue": "dataflow"},
         "govoplan.dataflow.dispatch_triggers": {"queue": "dataflow"},
         "govoplan.events.dispatch_outbox": {"queue": "events"},
         "govoplan.events.purge_outbox": {"queue": "events"},
@@ -60,6 +64,16 @@ celery.conf.update(
             "schedule": 60.0,
             "args": (100,),
         },
+        "dataflow-runs-every-five-seconds": {
+            "task": "govoplan.dataflow.dispatch_runs",
+            "schedule": 5.0,
+            "args": (10,),
+        },
+        "dataflow-run-retention-daily": {
+            "task": "govoplan.dataflow.purge_runs",
+            "schedule": 24 * 60 * 60.0,
+            "args": (500,),
+        },
         "platform-events-every-ten-seconds": {
             "task": "govoplan.events.dispatch_outbox",
             "schedule": 10.0,
@@ -131,6 +145,18 @@ def _dataflow_trigger_dispatcher(
     return capability
 
 
+def _dataflow_run_worker(
+    registry: PlatformRegistry | None = None,
+) -> DataflowRunWorker | None:
+    registry = registry or _platform_registry()
+    if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER):
+        return None
+    capability = registry.require_capability(CAPABILITY_DATAFLOW_RUN_WORKER)
+    if not isinstance(capability, DataflowRunWorker):
+        raise RuntimeError("Dataflow run worker capability is invalid")
+    return capability
+
+
 def _platform_event_outbox(
     registry: PlatformRegistry | None = None,
 ) -> PlatformEventOutbox | None:
@@ -234,6 +260,56 @@ def dispatch_dataflow_triggers(self, limit: int = 100):
         return result
 
 
+@celery.task(
+    name="govoplan.dataflow.dispatch_runs",
+    bind=True,
+    max_retries=0,
+)
+def dispatch_dataflow_runs(self, limit: int = 10):
+    """Claim and execute durable Dataflow runs outside the API process."""
+
+    from govoplan_core.db.session import get_database
+
+    with get_database().SessionLocal() as session:
+        provider = _dataflow_run_worker()
+        if provider is None:
+            return {
+                "claimed": 0,
+                "succeeded": 0,
+                "retrying": 0,
+                "failed": 0,
+                "cancelled": 0,
+            }
+        result = dict(
+            provider.dispatch_pending(
+                session,
+                limit=limit,
+                worker_id=getattr(self.request, "hostname", None),
+            )
+        )
+        session.commit()
+        return result
+
+
+@celery.task(
+    name="govoplan.dataflow.purge_runs",
+    bind=True,
+    max_retries=0,
+)
+def purge_dataflow_runs(self, limit: int = 500):
+    """Apply Dataflow evidence-retention policy without deleting run records."""
+
+    from govoplan_core.db.session import get_database
+
+    with get_database().SessionLocal() as session:
+        provider = _dataflow_run_worker()
+        if provider is None:
+            return {"purged": 0}
+        result = dict(provider.purge_expired(session, limit=limit))
+        session.commit()
+        return result
+
+
 @celery.task(
     name="govoplan.events.dispatch_outbox",
     bind=True,

src/govoplan_core/core/dataflows.py

@@ -10,6 +10,7 @@ from govoplan_core.core.events import PlatformEvent
 
 
 CAPABILITY_DATAFLOW_RUN_LIFECYCLE = "dataflow.runLifecycle"
+CAPABILITY_DATAFLOW_RUN_WORKER = "dataflow.runWorker"
 CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER = "dataflow.triggerDispatcher"
 
 
@@ -47,6 +48,10 @@ class DataflowRunRequest:
     revision: int
     idempotency_key: str
     row_limit: int = 500
+    execution_backend: str = "auto"
+    environment: str = "development"
+    max_attempts: int = 3
+    retention_days: int = 30
     publication: DataflowPublicationTarget | None = None
     invocation: AutomationInvocation = field(
         default_factory=AutomationInvocation
@@ -126,6 +131,34 @@ class DataflowTriggerDispatcher(Protocol):
         ...
 
 
+@runtime_checkable
+class DataflowRunWorker(Protocol):
+    """Durable worker boundary for queued Dataflow execution.
+
+    Implementations own claim transaction boundaries so a lease is committed
+    before potentially long-running execution starts.
+    """
+
+    def dispatch_pending(
+        self,
+        session: object,
+        *,
+        now: datetime | None = None,
+        limit: int = 10,
+        worker_id: str | None = None,
+    ) -> Mapping[str, object]:
+        ...
+
+    def purge_expired(
+        self,
+        session: object,
+        *,
+        now: datetime | None = None,
+        limit: int = 500,
+    ) -> Mapping[str, object]:
+        ...
+
+
 def dataflow_run_lifecycle(
     registry: object | None,
 ) -> DataflowRunLifecycleProvider | None:
@@ -133,6 +166,13 @@ def dataflow_run_lifecycle(
     return capability if isinstance(capability, DataflowRunLifecycleProvider) else None
 
 
+def dataflow_run_worker(
+    registry: object | None,
+) -> DataflowRunWorker | None:
+    capability = _capability(registry, CAPABILITY_DATAFLOW_RUN_WORKER)
+    return capability if isinstance(capability, DataflowRunWorker) else None
+
+
 def dataflow_trigger_dispatcher(
     registry: object | None,
 ) -> DataflowTriggerDispatcher | None:
@@ -157,6 +197,7 @@ def _capability(registry: object | None, name: str) -> object | None:
 
 __all__ = [
     "CAPABILITY_DATAFLOW_RUN_LIFECYCLE",
+    "CAPABILITY_DATAFLOW_RUN_WORKER",
     "CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER",
     "DataflowPublicationTarget",
     "DataflowRunConflictError",
@@ -166,7 +207,9 @@ __all__ = [
     "DataflowRunNotFoundError",
     "DataflowRunRequest",
     "DataflowRunUnavailableError",
+    "DataflowRunWorker",
     "DataflowTriggerDispatcher",
     "dataflow_run_lifecycle",
+    "dataflow_run_worker",
     "dataflow_trigger_dispatcher",
 ]