feat: preserve automation actor provenance

src/govoplan_dataflow/backend/schemas.py

@@ -429,6 +429,16 @@ class DataflowTriggerDeliveryListResponse(BaseModel):
     deliveries: list[DataflowTriggerDeliveryResponse]
 
 
+class DataflowEventActor(BaseModel):
+    type: str = Field(
+        min_length=1,
+        max_length=100,
+        pattern=r"^[A-Za-z0-9_.:-]+$",
+    )
+    id: str | None = Field(default=None, max_length=255)
+    label: str | None = Field(default=None, max_length=300)
+
+
 class DataflowEventDeliveryRequest(BaseModel):
     event_id: str = Field(min_length=1, max_length=128)
     type: str = Field(
@@ -445,6 +455,7 @@ class DataflowEventDeliveryRequest(BaseModel):
     occurred_at: datetime
     correlation_id: str | None = Field(default=None, max_length=128)
     causation_id: str | None = Field(default=None, max_length=128)
+    actor: DataflowEventActor | None = None
     classification: Literal[
         "public",
         "internal",

src/govoplan_dataflow/backend/triggers.py

@@ -746,9 +746,10 @@ def _dispatch_delivery(
             now=now,
             error="Access automation-principal capability is unavailable.",
         )
+    event = delivery.event_ or {}
     resolution = provider.resolve_automation_principal(
         session,
-        request=AutomationPrincipalRequest(
+        request=AutomationPrincipalRequest.delegated_user(
             tenant_id=trigger.tenant_id,
             account_id=trigger.authorization_account_id,
             membership_id=trigger.authorization_membership_id,
@@ -760,6 +761,8 @@ def _dispatch_delivery(
                 "delivery_ref": (
                     f"dataflow-trigger-delivery:{delivery.id}"
                 ),
+                "event_actor": event.get("actor"),
+                "operator_override": event.get("operator_override"),
             },
         ),
     )
@@ -783,7 +786,6 @@ def _dispatch_delivery(
             action="automate",
         )
         publication = _publication_target(trigger)
-        event = delivery.event_ or {}
         run, _ = start_pipeline_run(
             session,
             tenant_id=trigger.tenant_id,
@@ -813,6 +815,11 @@ def _dispatch_delivery(
                     causation_id=_optional_text(event.get("causation_id")),
                     scheduled_for=delivery.scheduled_for,
                     requested_by=trigger.authorization_account_id,
+                    metadata={
+                        "authorization_provenance": dict(
+                            resolution.provenance
+                        )
+                    },
                 ),
             ),
         )
@@ -968,6 +975,11 @@ class SqlDataflowTriggerDispatcher:
                 occurred_at=event.occurred_at,
                 correlation_id=event.correlation_id,
                 causation_id=event.causation_id,
+                actor=(
+                    event.actor.to_dict()
+                    if event.actor is not None
+                    else None
+                ),
                 classification=event.classification,
             ),
         )

tests/test_triggers.py

@@ -97,15 +97,20 @@ class CurrentPrincipalProvider:
     def __init__(self, actor: ApiPrincipal, *, allowed: bool = True) -> None:
         self.actor = actor
         self.allowed = allowed
+        self.last_request = None
 
     def resolve_automation_principal(self, _session, *, request):
+        self.last_request = request
         return AutomationPrincipalResolution(
             allowed=self.allowed,
             principal=self.actor if self.allowed else None,
             reason=None if self.allowed else "Owner authorization revoked.",
             granted_scopes=tuple(request.grant_scopes) if self.allowed else (),
             missing_scopes=() if self.allowed else tuple(request.grant_scopes),
-            provenance={"status": "current" if self.allowed else "revoked"},
+            provenance={
+                "status": "current" if self.allowed else "revoked",
+                "event_actor": request.context.get("event_actor"),
+            },
         )
 
 
@@ -116,12 +121,13 @@ class Registry:
         *,
         automation_allowed: bool = True,
     ) -> None:
+        self.automation_provider = CurrentPrincipalProvider(
+            actor,
+            allowed=automation_allowed,
+        )
         self.capabilities = {
             POLICY_CAPABILITY: AllowDefinitionPolicy(),
-            AUTOMATION_CAPABILITY: CurrentPrincipalProvider(
-                actor,
-                allowed=automation_allowed,
-            ),
+            AUTOMATION_CAPABILITY: self.automation_provider,
         }
 
     def has_capability(self, name: str) -> bool:
@@ -322,6 +328,11 @@ class DataflowTriggerTests(unittest.TestCase):
             module_id="files",
             payload={"folder_id": "incoming", "file_id": "file-1"},
             occurred_at=utcnow(),
+            actor={
+                "type": "user",
+                "id": "account-2",
+                "label": "Uploader",
+            },
         )
 
         self.assertEqual(
@@ -355,6 +366,22 @@ class DataflowTriggerTests(unittest.TestCase):
         self.assertEqual(1, result.succeeded)
         self.assertEqual("event", run.invocation_kind)
         self.assertEqual("event-1", run.request_["invocation"]["event_id"])
+        self.assertEqual(
+            {
+                "type": "user",
+                "id": "account-2",
+                "label": "Uploader",
+            },
+            self.registry.automation_provider.last_request.context[
+                "event_actor"
+            ],
+        )
+        self.assertEqual(
+            "account-2",
+            run.request_["invocation"]["metadata"][
+                "authorization_provenance"
+            ]["event_actor"]["id"],
+        )
 
     def test_revoked_owner_blocks_delivery_before_execution(self) -> None:
         trigger = self._event_trigger()