feat: harden governed runs and add dashboard integration

src/govoplan_dataflow/backend/governance.py

@@ -67,7 +67,7 @@ def normalize_definition_scope(
         return principal.tenant_id, "group", clean_id
     if clean_type == "user":
         if not clean_id:
-            clean_id = principal.membership_id or principal.account_id
+            clean_id = principal.account_id
         if (
             clean_id not in {principal.membership_id, principal.account_id}
             and not administrative
@@ -75,6 +75,8 @@ def normalize_definition_scope(
             raise PermissionError(
                 "Definitions can only be created for the current user."
             )
+        if clean_id == principal.membership_id:
+            clean_id = principal.account_id
         return principal.tenant_id, "user", clean_id
     raise ValueError(
         "Definition scope must be system, tenant, group, or user."

src/govoplan_dataflow/backend/manifest.py

@@ -7,6 +7,7 @@ from govoplan_core.core.module_guards import (
     persistent_table_uninstall_guard,
 )
 from govoplan_core.core.access import (
+    CAPABILITY_ACCESS_DIRECTORY,
     CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
 )
 from govoplan_core.core.dataflows import (
@@ -34,6 +35,7 @@ from govoplan_core.core.datasources import (
 from govoplan_core.core.policy import (
     CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
 )
+from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_dataflow.backend.db import models as dataflow_models
 
@@ -241,6 +243,7 @@ manifest = ModuleManifest(
         "workflow",
     ),
     optional_capabilities=(
+        CAPABILITY_ACCESS_DIRECTORY,
         CAPABILITY_DATASOURCE_CATALOGUE,
         CAPABILITY_DATASOURCE_LIFECYCLE,
         CAPABILITY_DATASOURCE_PUBLICATION,
@@ -320,6 +323,15 @@ manifest = ModuleManifest(
                 order=72,
             ),
         ),
+        view_surfaces=(
+            ViewSurface(
+                id="dataflow.widget.pipelines",
+                module_id=MODULE_ID,
+                kind="section",
+                label="Dataflows widget",
+                order=70,
+            ),
+        ),
     ),
     route_factory=_dataflow_router,
     capability_factories={

src/govoplan_dataflow/backend/router.py

@@ -1,8 +1,12 @@
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.orm import Session
 
+from govoplan_core.api.v1.schemas import (
+    ReferenceOptionListResponse,
+    ReferenceOptionResponse,
+)
 from govoplan_core.audit.logging import audit_event
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
 from govoplan_core.core.automation import AutomationInvocation
@@ -21,6 +25,11 @@ from govoplan_core.core.datasources import (
     datasource_catalogue,
     datasource_lifecycle,
 )
+from govoplan_core.core.references import (
+    access_scope_reference_options,
+    access_scope_reference_provider_available,
+    validate_access_scope_reference,
+)
 from govoplan_core.core.tabular_sources import (
     parse_tabular_csv,
 )
@@ -264,6 +273,37 @@ def api_node_types(
     return _node_library_response()
 
 
+@router.get("/scope-targets", response_model=ReferenceOptionListResponse)
+def api_scope_targets(
+    scope_type: str,
+    q: str = "",
+    selected: list[str] = Query(default=[]),
+    limit: int = Query(default=50, ge=1, le=200),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> ReferenceOptionListResponse:
+    _require_any_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE)
+    registry = get_registry()
+    try:
+        options = access_scope_reference_options(
+            registry,
+            principal,
+            scope_type=scope_type,
+            query=q,
+            selected_values=selected,
+            limit=limit,
+            administrative=has_scope(principal, ADMIN_SCOPE),
+        )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+    return ReferenceOptionListResponse(
+        options=[ReferenceOptionResponse(**item.to_dict()) for item in options],
+        provider_available=access_scope_reference_provider_available(registry),
+    )
+
+
 @router.get("/sources", response_model=TabularSourceListResponse)
 def api_list_sources(
     query: str = "",
@@ -417,6 +457,13 @@ def api_create_pipeline(
         payload = payload.model_copy(
             update={"scope_type": scope_type, "scope_id": scope_id}
         )
+        scope_id = validate_access_scope_reference(
+            get_registry(),
+            tenant_id=tenant_id or principal.tenant_id,
+            scope_type=scope_type,
+            scope_id=scope_id,
+        )
+        payload = payload.model_copy(update={"scope_id": scope_id})
         pipeline = create_pipeline(
             session,
             tenant_id=tenant_id or principal.tenant_id,
@@ -488,6 +535,26 @@ def api_update_pipeline(
             registry=get_registry(),
             action="edit",
         )
+        tenant_id, scope_type, scope_id = normalize_definition_scope(
+            principal,
+            scope_type=payload.scope_type,
+            scope_id=payload.scope_id,
+            administrative=has_scope(principal, ADMIN_SCOPE),
+        )
+        scope_id = validate_access_scope_reference(
+            get_registry(),
+            tenant_id=tenant_id or principal.tenant_id,
+            scope_type=scope_type,
+            scope_id=scope_id,
+            preserve_existing=(
+                existing.scope_id
+                if existing.scope_type == scope_type
+                else None
+            ),
+        )
+        payload = payload.model_copy(
+            update={"scope_type": scope_type, "scope_id": scope_id}
+        )
         pipeline = update_pipeline(
             session,
             tenant_id=principal.tenant_id,
@@ -495,7 +562,7 @@ def api_update_pipeline(
             actor_id=_actor_id(principal),
             payload=payload,
         )
-    except PermissionError as exc:
+    except (PermissionError, ValueError) as exc:
         raise _governance_http_error(exc) from exc
     except DataflowError as exc:
         raise _http_error(exc) from exc
@@ -579,6 +646,13 @@ def api_derive_pipeline(
         payload = payload.model_copy(
             update={"scope_type": scope_type, "scope_id": scope_id}
         )
+        scope_id = validate_access_scope_reference(
+            get_registry(),
+            tenant_id=tenant_id or principal.tenant_id,
+            scope_type=scope_type,
+            scope_id=scope_id,
+        )
+        payload = payload.model_copy(update={"scope_id": scope_id})
         pipeline = derive_pipeline(
             session,
             tenant_id=tenant_id or principal.tenant_id,

src/govoplan_dataflow/backend/triggers.py

@@ -707,7 +707,15 @@ def _dispatch_delivery(
     registry: object | None,
     now: datetime,
 ) -> str:
-    trigger = session.get(DataflowTrigger, delivery.trigger_id)
+    # Serialize capacity checks per trigger. Delivery rows are independently
+    # leased with SKIP LOCKED, so without this lock two workers could each
+    # claim a different delivery and both observe the same available slot.
+    trigger = session.scalar(
+        select(DataflowTrigger)
+        .where(DataflowTrigger.id == delivery.trigger_id)
+        .with_for_update()
+        .execution_options(populate_existing=True)
+    )
     pipeline = session.get(DataflowPipeline, delivery.pipeline_id)
     if trigger is None or pipeline is None:
         return _finish_delivery(