feat: version automation principal subjects

src/govoplan_core/core/automation.py

@@ -19,6 +19,8 @@ AutomationInvocationKind = Literal[
     "retry",
     "backfill",
 ]
+AutomationSubjectKind = Literal["delegated_user", "service_account"]
+AUTOMATION_PRINCIPAL_CONTRACT_VERSION = "1"
 
 
 @dataclass(frozen=True, slots=True)
@@ -38,11 +40,92 @@ class AutomationInvocation:
 @dataclass(frozen=True, slots=True)
 class AutomationPrincipalRequest:
     tenant_id: str
-    account_id: str
-    membership_id: str
     authorization_ref: str
     grant_scopes: tuple[str, ...]
+    account_id: str | None = None
+    membership_id: str | None = None
+    service_account_id: str | None = None
+    subject_kind: AutomationSubjectKind = "delegated_user"
     context: Mapping[str, object] = field(default_factory=dict)
+    contract_version: str = AUTOMATION_PRINCIPAL_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if (
+            self.contract_version
+            != AUTOMATION_PRINCIPAL_CONTRACT_VERSION
+        ):
+            raise ValueError(
+                "Unsupported automation-principal contract version"
+            )
+        if not self.tenant_id.strip():
+            raise ValueError("Automation tenant id is required")
+        if not self.authorization_ref.strip():
+            raise ValueError(
+                "Automation authorization artifact reference is required"
+            )
+        if self.subject_kind == "delegated_user":
+            if (
+                not self.account_id
+                or not self.membership_id
+                or self.service_account_id is not None
+            ):
+                raise ValueError(
+                    "Delegated-user automation requires account and "
+                    "membership references only"
+                )
+        elif (
+            not self.service_account_id
+            or self.account_id is not None
+            or self.membership_id is not None
+        ):
+            raise ValueError(
+                "Service-account automation requires only a service-account "
+                "reference"
+            )
+        if any(
+            not scope.strip()
+            for scope in self.grant_scopes
+        ):
+            raise ValueError("Automation grant scopes must not be empty")
+
+    @classmethod
+    def delegated_user(
+        cls,
+        *,
+        tenant_id: str,
+        account_id: str,
+        membership_id: str,
+        authorization_ref: str,
+        grant_scopes: tuple[str, ...],
+        context: Mapping[str, object] | None = None,
+    ) -> AutomationPrincipalRequest:
+        return cls(
+            tenant_id=tenant_id,
+            account_id=account_id,
+            membership_id=membership_id,
+            authorization_ref=authorization_ref,
+            grant_scopes=grant_scopes,
+            context=context or {},
+        )
+
+    @classmethod
+    def service_account(
+        cls,
+        *,
+        tenant_id: str,
+        service_account_id: str,
+        authorization_ref: str,
+        grant_scopes: tuple[str, ...],
+        context: Mapping[str, object] | None = None,
+    ) -> AutomationPrincipalRequest:
+        return cls(
+            tenant_id=tenant_id,
+            service_account_id=service_account_id,
+            subject_kind="service_account",
+            authorization_ref=authorization_ref,
+            grant_scopes=grant_scopes,
+            context=context or {},
+        )
 
 
 @dataclass(frozen=True, slots=True)
@@ -89,10 +172,12 @@ def automation_principal_provider(
 
 __all__ = [
     "CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER",
+    "AUTOMATION_PRINCIPAL_CONTRACT_VERSION",
     "AutomationInvocation",
     "AutomationInvocationKind",
     "AutomationPrincipalProvider",
     "AutomationPrincipalRequest",
     "AutomationPrincipalResolution",
+    "AutomationSubjectKind",
     "automation_principal_provider",
 ]

tests/test_automation_contract.py

@@ -25,6 +25,27 @@ class _Provider:
 
 
 class AutomationContractTests(unittest.TestCase):
+    def test_principal_request_has_explicit_subject_contracts(self) -> None:
+        service = AutomationPrincipalRequest.service_account(
+            tenant_id="tenant-1",
+            service_account_id="service-1",
+            authorization_ref="trigger:1",
+            grant_scopes=("dataflow:pipeline:run",),
+        )
+
+        self.assertEqual("service_account", service.subject_kind)
+        self.assertEqual("service-1", service.service_account_id)
+        self.assertIsNone(service.account_id)
+        with self.assertRaisesRegex(
+            ValueError,
+            "Delegated-user automation",
+        ):
+            AutomationPrincipalRequest(
+                tenant_id="tenant-1",
+                authorization_ref="trigger:1",
+                grant_scopes=("dataflow:pipeline:run",),
+            )
+
     def test_invocation_records_stable_trigger_provenance(self) -> None:
         invocation = AutomationInvocation(
             kind="event",