feat: manage bounded automation service accounts

src/govoplan_access/backend/api/v1/admin_schemas.py

@@ -876,6 +876,55 @@ class AdminApiKeyCreateResponse(ApiKeyAdminItem):
     secret: str
 
 
+class ServiceAccountItem(BaseModel):
+    id: str
+    tenant_id: str
+    name: str
+    description: str | None = None
+    scope_ceiling: list[str] = Field(default_factory=list)
+    is_active: bool
+    revision: int
+    created_by_account_id: str | None = None
+    updated_by_account_id: str | None = None
+    retired_at: datetime | None = None
+    created_at: datetime
+    updated_at: datetime
+
+
+class ServiceAccountListResponse(BaseModel):
+    items: list[ServiceAccountItem]
+
+
+class ServiceAccountCreateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    name: str = Field(min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=4000)
+    scope_ceiling: list[str] = Field(
+        default_factory=list,
+        max_length=200,
+    )
+
+
+class ServiceAccountUpdateRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    expected_revision: int = Field(ge=1)
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=4000)
+    scope_ceiling: list[str] | None = Field(
+        default=None,
+        max_length=200,
+    )
+    is_active: bool | None = None
+
+
+class ServiceAccountRetireRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    expected_revision: int = Field(ge=1)
+
+
 class AuditAdminItem(BaseModel):
     id: str
     scope: Literal["tenant", "system"] = "tenant"

src/govoplan_access/backend/api/v1/service_accounts.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.api.v1.admin_common import _resolve_tenant
+from govoplan_access.backend.api.v1.admin_schemas import (
+    ServiceAccountCreateRequest,
+    ServiceAccountItem,
+    ServiceAccountListResponse,
+    ServiceAccountRetireRequest,
+    ServiceAccountUpdateRequest,
+)
+from govoplan_access.backend.service_accounts import (
+    ServiceAccountConflictError,
+    ServiceAccountError,
+    ServiceAccountNotFoundError,
+    create_service_account,
+    get_service_account,
+    list_service_accounts,
+    retire_service_account,
+    update_service_account,
+)
+from govoplan_core.audit.logging import audit_from_principal
+from govoplan_core.auth import ApiPrincipal, require_scope
+from govoplan_core.db.session import get_session
+
+
+router = APIRouter(
+    prefix="/admin/service-accounts",
+    tags=["admin", "service-accounts"],
+)
+
+
+@router.get("", response_model=ServiceAccountListResponse)
+def list_managed_service_accounts(
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(
+        require_scope("access:service_account:read")
+    ),
+):
+    tenant = _resolve_tenant(session, principal, None)
+    return ServiceAccountListResponse(
+        items=[
+            _service_account_item(item)
+            for item in list_service_accounts(
+                session,
+                tenant_id=tenant.id,
+            )
+        ]
+    )
+
+
+@router.get(
+    "/{service_account_id}",
+    response_model=ServiceAccountItem,
+)
+def get_managed_service_account(
+    service_account_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(
+        require_scope("access:service_account:read")
+    ),
+):
+    tenant = _resolve_tenant(session, principal, None)
+    try:
+        item = get_service_account(
+            session,
+            tenant_id=tenant.id,
+            service_account_id=service_account_id,
+        )
+    except ServiceAccountError as exc:
+        raise _service_account_http_error(exc) from exc
+    return _service_account_item(item)
+
+
+@router.post(
+    "",
+    response_model=ServiceAccountItem,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_managed_service_account(
+    payload: ServiceAccountCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(
+        require_scope("access:service_account:write")
+    ),
+):
+    tenant = _resolve_tenant(session, principal, None)
+    try:
+        item = create_service_account(
+            session,
+            tenant=tenant,
+            principal=principal,
+            name=payload.name,
+            description=payload.description,
+            scope_ceiling=payload.scope_ceiling,
+        )
+    except (ServiceAccountError, PermissionError) as exc:
+        session.rollback()
+        raise _service_account_http_error(exc) from exc
+    audit_from_principal(
+        session,
+        principal,
+        action="service_account.created",
+        scope="tenant",
+        object_type="service_account",
+        object_id=item.id,
+        details={
+            "name": item.name,
+            "scope_ceiling": list(item.scope_ceiling),
+        },
+    )
+    session.commit()
+    return _service_account_item(item)
+
+
+@router.patch(
+    "/{service_account_id}",
+    response_model=ServiceAccountItem,
+)
+def update_managed_service_account(
+    service_account_id: str,
+    payload: ServiceAccountUpdateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(
+        require_scope("access:service_account:write")
+    ),
+):
+    tenant = _resolve_tenant(session, principal, None)
+    changes = {
+        field: getattr(payload, field)
+        for field in payload.model_fields_set
+        if field != "expected_revision"
+    }
+    for field in ("name", "scope_ceiling", "is_active"):
+        if field in changes and changes[field] is None:
+            raise HTTPException(
+                status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+                detail=f"{field} cannot be null",
+            )
+    try:
+        item = update_service_account(
+            session,
+            tenant_id=tenant.id,
+            service_account_id=service_account_id,
+            principal=principal,
+            expected_revision=payload.expected_revision,
+            changes=changes,
+        )
+    except (ServiceAccountError, PermissionError) as exc:
+        session.rollback()
+        raise _service_account_http_error(exc) from exc
+    audit_from_principal(
+        session,
+        principal,
+        action="service_account.updated",
+        scope="tenant",
+        object_type="service_account",
+        object_id=item.id,
+        details={
+            "changed_fields": sorted(changes),
+            "revision": item.revision,
+        },
+    )
+    session.commit()
+    return _service_account_item(item)
+
+
+@router.post(
+    "/{service_account_id}/retire",
+    response_model=ServiceAccountItem,
+)
+def retire_managed_service_account(
+    service_account_id: str,
+    payload: ServiceAccountRetireRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(
+        require_scope("access:service_account:write")
+    ),
+):
+    tenant = _resolve_tenant(session, principal, None)
+    try:
+        item = retire_service_account(
+            session,
+            tenant_id=tenant.id,
+            service_account_id=service_account_id,
+            principal=principal,
+            expected_revision=payload.expected_revision,
+        )
+    except (ServiceAccountError, PermissionError) as exc:
+        session.rollback()
+        raise _service_account_http_error(exc) from exc
+    audit_from_principal(
+        session,
+        principal,
+        action="service_account.retired",
+        scope="tenant",
+        object_type="service_account",
+        object_id=item.id,
+        details={"revision": item.revision},
+    )
+    session.commit()
+    return _service_account_item(item)
+
+
+def _service_account_item(item: object) -> ServiceAccountItem:
+    return ServiceAccountItem.model_validate(
+        item,
+        from_attributes=True,
+    )
+
+
+def _service_account_http_error(exc: Exception) -> HTTPException:
+    if isinstance(exc, ServiceAccountNotFoundError):
+        return HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        )
+    if isinstance(exc, ServiceAccountConflictError):
+        return HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        )
+    if isinstance(exc, PermissionError):
+        return HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=str(exc),
+        )
+    return HTTPException(
+        status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+        detail=str(exc),
+    )
+
+
+__all__ = ["router"]

src/govoplan_access/backend/auth/dependencies.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass, replace
 
 from fastapi import Depends, Header, HTTPException, Request, status
@@ -26,7 +27,15 @@ from govoplan_core.core.modules import AccessDecision
 from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.core.maintenance import MAINTENANCE_ACCESS_SCOPE, maintenance_response_detail, saved_maintenance_mode
 from govoplan_core.db.session import get_database, get_session
-from govoplan_access.backend.db.models import Account, ApiKey, AuthSession, Role, Tenant, User
+from govoplan_access.backend.db.models import (
+    Account,
+    ApiKey,
+    AuthSession,
+    Role,
+    ServiceAccount,
+    Tenant,
+    User,
+)
 from govoplan_access.backend.semantic import collect_external_function_roles, identity_id_for_account
 from govoplan_access.backend.security.api_keys import authenticate_api_key
 from govoplan_access.backend.security.sessions import (
@@ -605,6 +614,28 @@ class AccessAutomationPrincipalProvider:
                 "Access automation principal resolution requires a "
                 "SQLAlchemy session"
             )
+        if request.subject_kind == "service_account":
+            return _resolve_service_account_automation(
+                session,
+                request=request,
+            )
+        return _resolve_delegated_user_automation(
+            session,
+            request=request,
+            idm_directory=self._idm_directory,
+            identity_directory=self._identity_directory,
+            organization_directory=self._organization_directory,
+        )
+
+
+def _resolve_delegated_user_automation(
+    session: Session,
+    *,
+    request: AutomationPrincipalRequest,
+    idm_directory: IdmDirectory | None,
+    identity_directory: IdentityDirectory | None,
+    organization_directory: OrganizationDirectory | None,
+) -> AutomationPrincipalResolution:
     account = session.get(Account, request.account_id)
     user = session.get(User, request.membership_id)
     tenant = session.get(Tenant, request.tenant_id)
@@ -618,29 +649,21 @@ class AccessAutomationPrincipalProvider:
         or user.account_id != account.id
         or user.tenant_id != tenant.id
     ):
-            return AutomationPrincipalResolution(
+        return _automation_denied(
-                allowed=False,
+            request,
+            status="inactive_or_inconsistent",
             reason=(
                 "The automation owner is inactive, missing, or no longer "
                 "belongs to the tenant."
             ),
-                missing_scopes=tuple(sorted(set(request.grant_scopes))),
-                provenance={
-                    "authorization_ref": request.authorization_ref,
-                    "tenant_id": request.tenant_id,
-                    "account_id": request.account_id,
-                    "membership_id": request.membership_id,
-                    "status": "inactive_or_inconsistent",
-                },
         )
-
     idm_assignments, idm_roles = _principal_idm_context(
         session,
         user=user,
         account=account,
         tenant_id=request.tenant_id,
-            idm_directory=self._idm_directory,
+        idm_directory=idm_directory,
-            organization_directory=self._organization_directory,
+        organization_directory=organization_directory,
     )
     authorization_context = collect_user_authorization_context(
         session,
@@ -649,37 +672,21 @@ class AccessAutomationPrincipalProvider:
         include_system=False,
         extra_roles=idm_roles,
     )
-        current_scopes = authorization_context.scopes
+    granted_scopes, missing_scopes = _current_trigger_grants(
-        granted_scopes = tuple(
+        request.grant_scopes,
-            sorted(
+        current_scopes=authorization_context.scopes,
-                {
-                    required
-                    for required in request.grant_scopes
-                    if scopes_grant_compatible(current_scopes, required)
-                }
-            )
-        )
-        missing_scopes = tuple(
-            sorted(set(request.grant_scopes) - set(granted_scopes))
     )
     if missing_scopes:
-            return AutomationPrincipalResolution(
+        return _automation_denied(
-                allowed=False,
+            request,
+            status="authorization_reduced",
             reason=(
                 "The automation owner no longer has every scope granted "
                 "to this trigger."
             ),
             granted_scopes=granted_scopes,
             missing_scopes=missing_scopes,
-                provenance={
-                    "authorization_ref": request.authorization_ref,
-                    "tenant_id": request.tenant_id,
-                    "account_id": request.account_id,
-                    "membership_id": request.membership_id,
-                    "status": "authorization_reduced",
-                },
         )
-
     principal_ref = _build_principal_ref(
         session,
         account=account,
@@ -688,39 +695,260 @@ class AccessAutomationPrincipalProvider:
         scopes=list(granted_scopes),
         auth_method="service_account",
         idm_assignments=idm_assignments,
-            identity_directory=self._identity_directory,
+        identity_directory=identity_directory,
         extra_roles=idm_roles,
         authorization_context=authorization_context,
         include_system_roles=False,
     )
     principal_ref = replace(
         principal_ref,
-            service_account_id=request.authorization_ref,
+        service_account_id=None,
         acting_for_account_id=account.id,
     )
+    return _automation_allowed(
+        session,
+        request=request,
+        principal_ref=principal_ref,
+        granted_scopes=granted_scopes,
+    )
+
+
+def _resolve_service_account_automation(
+    session: Session,
+    *,
+    request: AutomationPrincipalRequest,
+) -> AutomationPrincipalResolution:
+    item = session.get(ServiceAccount, request.service_account_id)
+    tenant = session.get(Tenant, request.tenant_id)
+    account = (
+        session.get(Account, item.account_id)
+        if item is not None
+        else None
+    )
+    user = (
+        session.get(User, item.membership_id)
+        if item is not None
+        else None
+    )
+    if (
+        item is None
+        or tenant is None
+        or account is None
+        or user is None
+        or item.tenant_id != request.tenant_id
+        or not item.is_active
+        or not tenant.is_active
+        or not account.is_active
+        or not user.is_active
+        or item.account_id != account.id
+        or item.membership_id != user.id
+        or user.account_id != account.id
+        or user.tenant_id != tenant.id
+        or account.auth_provider != "service_account"
+        or user.auth_provider != "service_account"
+    ):
+        return _automation_denied(
+            request,
+            status="inactive_or_inconsistent",
+            reason=(
+                "The service account is inactive, missing, or no longer "
+                "belongs to the tenant."
+            ),
+        )
+    granted_scopes, missing_scopes = _current_trigger_grants(
+        request.grant_scopes,
+        current_scopes=item.scope_ceiling,
+    )
+    if missing_scopes:
+        return _automation_denied(
+            request,
+            status="authorization_reduced",
+            reason=(
+                "The service account no longer has every scope granted "
+                "to this trigger."
+            ),
+            granted_scopes=granted_scopes,
+            missing_scopes=missing_scopes,
+        )
+    principal_ref = PrincipalRef(
+        account_id=account.id,
+        membership_id=user.id,
+        tenant_id=tenant.id,
+        scopes=frozenset(granted_scopes),
+        auth_method="service_account",
+        service_account_id=item.id,
+        email=None,
+        display_name=item.name,
+    )
+    return _automation_allowed(
+        session,
+        request=request,
+        principal_ref=principal_ref,
+        granted_scopes=granted_scopes,
+    )
+
+
+def _current_trigger_grants(
+    trigger_scopes: tuple[str, ...],
+    *,
+    current_scopes: list[str] | tuple[str, ...],
+) -> tuple[tuple[str, ...], tuple[str, ...]]:
+    granted = tuple(
+        sorted(
+            {
+                required
+                for required in trigger_scopes
+                if scopes_grant_compatible(
+                    current_scopes,
+                    required,
+                )
+            }
+        )
+    )
+    missing = tuple(
+        sorted(set(trigger_scopes) - set(granted))
+    )
+    return granted, missing
+
+
+def _automation_allowed(
+    session: Session,
+    *,
+    request: AutomationPrincipalRequest,
+    principal_ref: PrincipalRef,
+    granted_scopes: tuple[str, ...],
+) -> AutomationPrincipalResolution:
     api_principal = _api_principal_from_ref(
         session,
         principal_ref,
         permission_evaluator=LegacyPermissionEvaluator(),
     )
-        return AutomationPrincipalResolution(
+    provenance = _automation_provenance(
-            allowed=True,
+        request,
-            principal=api_principal,
+        status="current_authorization_resolved",
-            granted_scopes=granted_scopes,
+        current_principal={
-            provenance={
+            "kind": request.subject_kind,
-                "authorization_ref": request.authorization_ref,
+            "account_id": principal_ref.account_id,
-                "tenant_id": request.tenant_id,
+            "membership_id": principal_ref.membership_id,
-                "account_id": request.account_id,
+            "service_account_id": principal_ref.service_account_id,
-                "membership_id": request.membership_id,
             "role_ids": sorted(principal_ref.role_ids),
             "group_ids": sorted(principal_ref.group_ids),
             "function_assignment_ids": sorted(
                 principal_ref.function_assignment_ids
             ),
-                "delegation_ids": sorted(principal_ref.delegation_ids),
+            "delegation_ids": sorted(
-                "status": "current_authorization_resolved",
+                principal_ref.delegation_ids
+            ),
+            "granted_scopes": list(granted_scopes),
         },
     )
+    return AutomationPrincipalResolution(
+        allowed=True,
+        principal=api_principal,
+        granted_scopes=granted_scopes,
+        provenance=provenance,
+    )
+
+
+def _automation_denied(
+    request: AutomationPrincipalRequest,
+    *,
+    status: str,
+    reason: str,
+    granted_scopes: tuple[str, ...] = (),
+    missing_scopes: tuple[str, ...] | None = None,
+) -> AutomationPrincipalResolution:
+    missing = (
+        tuple(sorted(set(request.grant_scopes)))
+        if missing_scopes is None
+        else missing_scopes
+    )
+    return AutomationPrincipalResolution(
+        allowed=False,
+        reason=reason,
+        granted_scopes=granted_scopes,
+        missing_scopes=missing,
+        provenance=_automation_provenance(
+            request,
+            status=status,
+            current_principal=None,
+        ),
+    )
+
+
+def _automation_provenance(
+    request: AutomationPrincipalRequest,
+    *,
+    status: str,
+    current_principal: Mapping[str, object] | None,
+) -> dict[str, object]:
+    return {
+        "contract_version": request.contract_version,
+        "authorization_artifact": {
+            "ref": request.authorization_ref,
+        },
+        "trigger_owner": _automation_trigger_owner(request),
+        "current_automation_principal": (
+            dict(current_principal)
+            if current_principal is not None
+            else None
+        ),
+        "event_actor": _automation_context_actor(
+            request.context.get("event_actor")
+        ),
+        "operator_override": _automation_context_actor(
+            request.context.get("operator_override")
+        ),
+        "trigger_ref": _optional_context_text(
+            request.context.get("trigger_ref")
+        ),
+        "delivery_ref": _optional_context_text(
+            request.context.get("delivery_ref")
+        ),
+        "status": status,
+    }
+
+
+def _automation_trigger_owner(
+    request: AutomationPrincipalRequest,
+) -> dict[str, object]:
+    return {
+        "kind": request.subject_kind,
+        "tenant_id": request.tenant_id,
+        "account_id": request.account_id,
+        "membership_id": request.membership_id,
+        "service_account_id": request.service_account_id,
+    }
+
+
+def _automation_context_actor(
+    value: object,
+) -> dict[str, str] | None:
+    if not isinstance(value, Mapping):
+        return None
+    result = {
+        key: str(value[key]).strip()
+        for key in (
+            "kind",
+            "type",
+            "id",
+            "label",
+            "account_id",
+            "membership_id",
+            "service_account_id",
+            "reason",
+        )
+        if value.get(key) is not None
+        and str(value[key]).strip()
+    }
+    return result or None
+
+
+def _optional_context_text(value: object) -> str | None:
+    if value is None:
+        return None
+    clean = str(value).strip()
+    return clean[:300] or None
 
 
 def get_api_principal(

src/govoplan_access/backend/db/models.py

@@ -4,7 +4,18 @@ import uuid
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Index, String, Text, UniqueConstraint, JSON, text
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    text,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from govoplan_access.backend.db.base import AccessBase, TimestampMixin
@@ -110,6 +121,91 @@ class User(AccessBase, TimestampMixin):
     auth_sessions: Mapped[list[AuthSession]] = relationship(back_populates="user", cascade="all, delete-orphan")
 
 
+class ServiceAccount(AccessBase, TimestampMixin):
+    """Managed non-login principal for current-authority automation."""
+
+    __tablename__ = "access_service_accounts"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "normalized_name",
+            name="uq_access_service_accounts_tenant_name",
+        ),
+        UniqueConstraint(
+            "account_id",
+            name="uq_access_service_accounts_account",
+        ),
+        UniqueConstraint(
+            "membership_id",
+            name="uq_access_service_accounts_membership",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        default=new_uuid,
+    )
+    tenant_id: Mapped[str] = mapped_column(
+        String(36),
+        nullable=False,
+        index=True,
+    )
+    account_id: Mapped[str] = mapped_column(
+        ForeignKey("access_accounts.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    membership_id: Mapped[str] = mapped_column(
+        ForeignKey("access_users.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    name: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+    )
+    normalized_name: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+    )
+    description: Mapped[str | None] = mapped_column(Text)
+    scope_ceiling: Mapped[list[str]] = mapped_column(
+        JSON,
+        default=list,
+        nullable=False,
+    )
+    is_active: Mapped[bool] = mapped_column(
+        Boolean,
+        default=True,
+        nullable=False,
+        index=True,
+    )
+    revision: Mapped[int] = mapped_column(
+        Integer,
+        default=1,
+        nullable=False,
+    )
+    created_by_account_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_accounts.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    updated_by_account_id: Mapped[str | None] = mapped_column(
+        ForeignKey("access_accounts.id", ondelete="SET NULL"),
+        nullable=True,
+    )
+    retired_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+        index=True,
+    )
+    settings: Mapped[dict[str, Any]] = mapped_column(
+        JSON,
+        default=dict,
+        nullable=False,
+    )
+
+
 class Group(AccessBase, TimestampMixin):
     __tablename__ = "access_groups"
     __table_args__ = (UniqueConstraint("tenant_id", "slug", name="uq_groups_tenant_slug"),)
@@ -358,6 +454,7 @@ __all__ = [
     "IdentityAccountLink",
     "OrganizationUnit",
     "Role",
+    "ServiceAccount",
     "SystemRoleAssignment",
     "Tenant",
     "User",

src/govoplan_access/backend/manifest.py

@@ -93,6 +93,8 @@ ACCESS_PERMISSIONS: tuple[PermissionDefinition, ...] = (
     _permission("access:api_key:read", "View API keys", "List API keys without revealing secrets.", "Tenant access", "tenant"),
     _permission("access:api_key:create", "Create API keys", "Create tenant API keys within delegation limits.", "Tenant access", "tenant"),
     _permission("access:api_key:revoke", "Revoke API keys", "Revoke tenant API keys.", "Tenant access", "tenant"),
+    _permission("access:service_account:read", "View service accounts", "List non-login automation principals and their current scope ceilings.", "Tenant access", "tenant"),
+    _permission("access:service_account:write", "Manage service accounts", "Create, update, suspend, and retire scope-bounded automation principals.", "Tenant access", "tenant"),
     _permission("access:setting:read", "View settings", "Read access and governance settings.", "Tenant access", "tenant"),
     _permission("access:setting:write", "Manage settings", "Update access and governance settings.", "Tenant access", "tenant"),
     _permission("access:credential:read", "View credentials", "List reusable credential envelopes without revealing secret values.", "Tenant access", "tenant"),
@@ -192,6 +194,8 @@ ACCESS_ROLE_TEMPLATES: tuple[RoleTemplate, ...] = (
             "access:api_key:read",
             "access:api_key:create",
             "access:api_key:revoke",
+            "access:service_account:read",
+            "access:service_account:write",
             "access:setting:read",
             "access:setting:write",
             "access:credential:read",
@@ -244,6 +248,7 @@ ADMIN_READ_SCOPES = (
     "admin:groups:read",
     "admin:roles:read",
     "admin:api_keys:read",
+    "access:service_account:read",
     "admin:settings:read",
     "system:tenants:read",
     "system:accounts:read",
@@ -619,10 +624,14 @@ def _route_factory(context: ModuleContext):
 
     from govoplan_access.backend.api.v1.auth import router as auth_router
     from govoplan_access.backend.api.v1.routes import router as access_admin_router
+    from govoplan_access.backend.api.v1.service_accounts import (
+        router as service_account_router,
+    )
 
     router = APIRouter()
     router.include_router(auth_router)
     router.include_router(access_admin_router)
+    router.include_router(service_account_router)
     return router
 
 
@@ -641,7 +650,7 @@ manifest = ModuleManifest(
         ModuleInterfaceProvider(name=CAPABILITY_ACCESS_PEOPLE_SEARCH, version="0.1.0"),
         ModuleInterfaceProvider(
             name="auth.automation_principal",
-            version="0.1.0",
+            version="0.2.0",
         ),
     ),
     permissions=ACCESS_PERMISSIONS,
@@ -660,6 +669,7 @@ manifest = ModuleManifest(
             access_models.User,
             access_models.Group,
             access_models.Role,
+            access_models.ServiceAccount,
             access_models.OrganizationUnit,
             access_models.Function,
             access_models.FunctionRoleAssignment,

src/govoplan_access/backend/migrations/dev_versions/b6d9f2a5c8e1_service_accounts.py

@@ -0,0 +1,141 @@
+"""managed automation service accounts
+
+Revision ID: b6d9f2a5c8e1
+Revises: 4a5b6c7d8e9f
+Create Date: 2026-07-29 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b6d9f2a5c8e1"
+down_revision = "4a5b6c7d8e9f"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    if (
+        "access_service_accounts"
+        in sa.inspect(op.get_bind()).get_table_names()
+    ):
+        return
+    op.create_table(
+        "access_service_accounts",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("account_id", sa.String(length=36), nullable=False),
+        sa.Column("membership_id", sa.String(length=36), nullable=False),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("normalized_name", sa.String(length=255), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("scope_ceiling", sa.JSON(), nullable=False),
+        sa.Column("is_active", sa.Boolean(), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column(
+            "created_by_account_id",
+            sa.String(length=36),
+            nullable=True,
+        ),
+        sa.Column(
+            "updated_by_account_id",
+            sa.String(length=36),
+            nullable=True,
+        ),
+        sa.Column("retired_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("settings", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["account_id"],
+            ["access_accounts.id"],
+            name=op.f(
+                "fk_access_service_accounts_account_id_access_accounts"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["created_by_account_id"],
+            ["access_accounts.id"],
+            name=op.f(
+                "fk_access_service_accounts_created_by_account_id_"
+                "access_accounts"
+            ),
+            ondelete="SET NULL",
+        ),
+        sa.ForeignKeyConstraint(
+            ["membership_id"],
+            ["access_users.id"],
+            name=op.f(
+                "fk_access_service_accounts_membership_id_access_users"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["tenant_id"],
+            ["core_scopes.id"],
+            name=op.f(
+                "fk_access_service_accounts_tenant_id_core_scopes"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["updated_by_account_id"],
+            ["access_accounts.id"],
+            name=op.f(
+                "fk_access_service_accounts_updated_by_account_id_"
+                "access_accounts"
+            ),
+            ondelete="SET NULL",
+        ),
+        sa.PrimaryKeyConstraint(
+            "id",
+            name=op.f("pk_access_service_accounts"),
+        ),
+        sa.UniqueConstraint(
+            "account_id",
+            name="uq_access_service_accounts_account",
+        ),
+        sa.UniqueConstraint(
+            "membership_id",
+            name="uq_access_service_accounts_membership",
+        ),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "normalized_name",
+            name="uq_access_service_accounts_tenant_name",
+        ),
+    )
+    for name, columns in (
+        (
+            "ix_access_service_accounts_account_id",
+            ["account_id"],
+        ),
+        (
+            "ix_access_service_accounts_is_active",
+            ["is_active"],
+        ),
+        (
+            "ix_access_service_accounts_membership_id",
+            ["membership_id"],
+        ),
+        (
+            "ix_access_service_accounts_retired_at",
+            ["retired_at"],
+        ),
+        (
+            "ix_access_service_accounts_tenant_id",
+            ["tenant_id"],
+        ),
+    ):
+        op.create_index(name, "access_service_accounts", columns)
+
+
+def downgrade() -> None:
+    if (
+        "access_service_accounts"
+        in sa.inspect(op.get_bind()).get_table_names()
+    ):
+        op.drop_table("access_service_accounts")

src/govoplan_access/backend/migrations/versions/b6d9f2a5c8e1_v0112_service_accounts.py

@@ -0,0 +1,141 @@
+"""managed automation service accounts
+
+Revision ID: b6d9f2a5c8e1
+Revises: 4a5b6c7d8e9f
+Create Date: 2026-07-29 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b6d9f2a5c8e1"
+down_revision = "4a5b6c7d8e9f"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    if (
+        "access_service_accounts"
+        in sa.inspect(op.get_bind()).get_table_names()
+    ):
+        return
+    op.create_table(
+        "access_service_accounts",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("account_id", sa.String(length=36), nullable=False),
+        sa.Column("membership_id", sa.String(length=36), nullable=False),
+        sa.Column("name", sa.String(length=255), nullable=False),
+        sa.Column("normalized_name", sa.String(length=255), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("scope_ceiling", sa.JSON(), nullable=False),
+        sa.Column("is_active", sa.Boolean(), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column(
+            "created_by_account_id",
+            sa.String(length=36),
+            nullable=True,
+        ),
+        sa.Column(
+            "updated_by_account_id",
+            sa.String(length=36),
+            nullable=True,
+        ),
+        sa.Column("retired_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("settings", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["account_id"],
+            ["access_accounts.id"],
+            name=op.f(
+                "fk_access_service_accounts_account_id_access_accounts"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["created_by_account_id"],
+            ["access_accounts.id"],
+            name=op.f(
+                "fk_access_service_accounts_created_by_account_id_"
+                "access_accounts"
+            ),
+            ondelete="SET NULL",
+        ),
+        sa.ForeignKeyConstraint(
+            ["membership_id"],
+            ["access_users.id"],
+            name=op.f(
+                "fk_access_service_accounts_membership_id_access_users"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["tenant_id"],
+            ["core_scopes.id"],
+            name=op.f(
+                "fk_access_service_accounts_tenant_id_core_scopes"
+            ),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["updated_by_account_id"],
+            ["access_accounts.id"],
+            name=op.f(
+                "fk_access_service_accounts_updated_by_account_id_"
+                "access_accounts"
+            ),
+            ondelete="SET NULL",
+        ),
+        sa.PrimaryKeyConstraint(
+            "id",
+            name=op.f("pk_access_service_accounts"),
+        ),
+        sa.UniqueConstraint(
+            "account_id",
+            name="uq_access_service_accounts_account",
+        ),
+        sa.UniqueConstraint(
+            "membership_id",
+            name="uq_access_service_accounts_membership",
+        ),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "normalized_name",
+            name="uq_access_service_accounts_tenant_name",
+        ),
+    )
+    for name, columns in (
+        (
+            "ix_access_service_accounts_account_id",
+            ["account_id"],
+        ),
+        (
+            "ix_access_service_accounts_is_active",
+            ["is_active"],
+        ),
+        (
+            "ix_access_service_accounts_membership_id",
+            ["membership_id"],
+        ),
+        (
+            "ix_access_service_accounts_retired_at",
+            ["retired_at"],
+        ),
+        (
+            "ix_access_service_accounts_tenant_id",
+            ["tenant_id"],
+        ),
+    ):
+        op.create_index(name, "access_service_accounts", columns)
+
+
+def downgrade() -> None:
+    if (
+        "access_service_accounts"
+        in sa.inspect(op.get_bind()).get_table_names()
+    ):
+        op.drop_table("access_service_accounts")

src/govoplan_access/backend/service_accounts.py

@@ -0,0 +1,283 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db.base import utcnow
+from govoplan_access.backend.db.models import (
+    Account,
+    ServiceAccount,
+    Tenant,
+    User,
+    new_uuid,
+)
+from govoplan_core.auth import ApiPrincipal
+
+
+class ServiceAccountError(ValueError):
+    pass
+
+
+class ServiceAccountNotFoundError(ServiceAccountError):
+    pass
+
+
+class ServiceAccountConflictError(ServiceAccountError):
+    pass
+
+
+def list_service_accounts(
+    session: Session,
+    *,
+    tenant_id: str,
+) -> list[ServiceAccount]:
+    return list(
+        session.scalars(
+            select(ServiceAccount)
+            .where(ServiceAccount.tenant_id == tenant_id)
+            .order_by(
+                ServiceAccount.normalized_name,
+                ServiceAccount.id,
+            )
+        )
+    )
+
+
+def get_service_account(
+    session: Session,
+    *,
+    tenant_id: str,
+    service_account_id: str,
+    lock: bool = False,
+) -> ServiceAccount:
+    query = select(ServiceAccount).where(
+        ServiceAccount.id == service_account_id,
+        ServiceAccount.tenant_id == tenant_id,
+    )
+    if lock:
+        query = query.with_for_update()
+    item = session.scalar(query)
+    if item is None:
+        raise ServiceAccountNotFoundError(
+            "Service account was not found"
+        )
+    return item
+
+
+def create_service_account(
+    session: Session,
+    *,
+    tenant: Tenant,
+    principal: ApiPrincipal,
+    name: str,
+    description: str | None,
+    scope_ceiling: Iterable[str],
+) -> ServiceAccount:
+    clean_name = _service_account_name(name)
+    scopes = _service_account_scopes(
+        principal,
+        scope_ceiling,
+    )
+    service_account_id = new_uuid()
+    internal_email = (
+        f"service-account-{service_account_id}@govoplan.invalid"
+    )
+    account = Account(
+        id=new_uuid(),
+        email=internal_email,
+        normalized_email=internal_email,
+        display_name=clean_name,
+        is_active=True,
+        auth_provider="service_account",
+        password_hash=None,
+        password_reset_required=False,
+    )
+    membership = User(
+        id=new_uuid(),
+        tenant_id=tenant.id,
+        account=account,
+        email=internal_email,
+        display_name=clean_name,
+        is_active=True,
+        is_tenant_admin=False,
+        auth_provider="service_account",
+        password_hash=None,
+        settings={"managed_service_account": service_account_id},
+    )
+    item = ServiceAccount(
+        id=service_account_id,
+        tenant_id=tenant.id,
+        account_id=account.id,
+        membership_id=membership.id,
+        name=clean_name,
+        normalized_name=_normalized_name(clean_name),
+        description=_optional_text(description),
+        scope_ceiling=list(scopes),
+        is_active=True,
+        revision=1,
+        created_by_account_id=principal.account_id,
+        updated_by_account_id=principal.account_id,
+        settings={},
+    )
+    session.add_all((account, membership, item))
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        raise ServiceAccountConflictError(
+            "A service account with this name already exists"
+        ) from exc
+    return item
+
+
+def update_service_account(
+    session: Session,
+    *,
+    tenant_id: str,
+    service_account_id: str,
+    principal: ApiPrincipal,
+    expected_revision: int,
+    changes: Mapping[str, object],
+) -> ServiceAccount:
+    item = get_service_account(
+        session,
+        tenant_id=tenant_id,
+        service_account_id=service_account_id,
+        lock=True,
+    )
+    if item.revision != expected_revision:
+        raise ServiceAccountConflictError(
+            "Service account changed on the server; reload before saving"
+        )
+    account = session.get(Account, item.account_id)
+    membership = session.get(User, item.membership_id)
+    if account is None or membership is None:
+        raise ServiceAccountConflictError(
+            "Service account backing identity is missing"
+        )
+    if "name" in changes:
+        clean_name = _service_account_name(str(changes["name"]))
+        item.name = clean_name
+        item.normalized_name = _normalized_name(clean_name)
+        account.display_name = clean_name
+        membership.display_name = clean_name
+    if "description" in changes:
+        value = changes["description"]
+        item.description = _optional_text(
+            str(value) if value is not None else None
+        )
+    if "scope_ceiling" in changes:
+        raw_scopes = changes["scope_ceiling"]
+        if not isinstance(raw_scopes, Iterable) or isinstance(
+            raw_scopes,
+            (str, bytes),
+        ):
+            raise ServiceAccountError(
+                "Service account scope ceiling is invalid"
+            )
+        item.scope_ceiling = list(
+            _service_account_scopes(
+                principal,
+                (str(scope) for scope in raw_scopes),
+            )
+        )
+    if "is_active" in changes:
+        active = bool(changes["is_active"])
+        item.is_active = active
+        account.is_active = active
+        membership.is_active = active
+        item.retired_at = None if active else utcnow()
+    item.revision += 1
+    item.updated_by_account_id = principal.account_id
+    try:
+        session.flush()
+    except IntegrityError as exc:
+        raise ServiceAccountConflictError(
+            "A service account with this name already exists"
+        ) from exc
+    return item
+
+
+def retire_service_account(
+    session: Session,
+    *,
+    tenant_id: str,
+    service_account_id: str,
+    principal: ApiPrincipal,
+    expected_revision: int,
+) -> ServiceAccount:
+    return update_service_account(
+        session,
+        tenant_id=tenant_id,
+        service_account_id=service_account_id,
+        principal=principal,
+        expected_revision=expected_revision,
+        changes={"is_active": False},
+    )
+
+
+def _service_account_name(value: str) -> str:
+    clean = " ".join(value.split())
+    if not 1 <= len(clean) <= 255:
+        raise ServiceAccountError(
+            "Service account name must contain between 1 and 255 characters"
+        )
+    return clean
+
+
+def _normalized_name(value: str) -> str:
+    return value.casefold()
+
+
+def _optional_text(value: str | None) -> str | None:
+    if value is None:
+        return None
+    clean = value.strip()
+    if len(clean) > 4000:
+        raise ServiceAccountError(
+            "Service account description is too long"
+        )
+    return clean or None
+
+
+def _service_account_scopes(
+    principal: ApiPrincipal,
+    values: Iterable[str],
+) -> tuple[str, ...]:
+    scopes = tuple(
+        sorted(
+            {
+                str(value).strip()
+                for value in values
+                if str(value).strip()
+            }
+        )
+    )
+    if len(scopes) > 200:
+        raise ServiceAccountError(
+            "Service accounts support at most 200 scope grants"
+        )
+    denied = tuple(
+        scope for scope in scopes
+        if not principal.has(scope)
+    )
+    if denied:
+        raise PermissionError(
+            "Cannot grant service-account scopes outside the current "
+            f"administrator authority: {', '.join(denied)}"
+        )
+    return scopes
+
+
+__all__ = [
+    "ServiceAccountConflictError",
+    "ServiceAccountError",
+    "ServiceAccountNotFoundError",
+    "create_service_account",
+    "get_service_account",
+    "list_service_accounts",
+    "retire_service_account",
+    "update_service_account",
+]

tests/test_automation_principal.py

@@ -10,7 +10,11 @@ from govoplan_access.backend.auth.dependencies import (
     AccessAutomationPrincipalProvider,
 )
 from govoplan_access.backend.db.base import AccessBase
-from govoplan_access.backend.db.models import Account, User
+from govoplan_access.backend.db.models import (
+    Account,
+    ServiceAccount,
+    User,
+)
 from govoplan_access.backend.manifest import manifest
 from govoplan_access.backend.security.sessions import (
     UserAuthorizationContext,
@@ -70,6 +74,19 @@ class AutomationPrincipalTests(unittest.TestCase):
                 "dataflow:pipeline:run",
                 "datasources:catalogue:read",
             ),
+            context={
+                "trigger_ref": "dataflow-trigger:1",
+                "delivery_ref": "dataflow-delivery:1",
+                "event_actor": {
+                    "type": "user",
+                    "id": "event-user-1",
+                },
+                "operator_override": {
+                    "type": "user",
+                    "id": "operator-1",
+                    "reason": "approved replay",
+                },
+            },
         )
 
     def test_resolution_intersects_trigger_grant_with_current_scopes(self) -> None:
@@ -106,14 +123,35 @@ class AutomationPrincipalTests(unittest.TestCase):
             ),
             result.principal.scopes,
         )
+        self.assertIsNone(
+            result.principal.principal.service_account_id
+        )
         self.assertEqual(
-            "dataflow-trigger:1",
+            self.account.id,
-            result.principal.principal.service_account_id,
+            result.principal.principal.acting_for_account_id,
         )
         self.assertNotIn(
             "system:settings:write",
             result.principal.scopes,
         )
+        self.assertEqual(
+            "delegated_user",
+            result.provenance["trigger_owner"]["kind"],
+        )
+        self.assertEqual(
+            "event-user-1",
+            result.provenance["event_actor"]["id"],
+        )
+        self.assertEqual(
+            "operator-1",
+            result.provenance["operator_override"]["id"],
+        )
+        self.assertEqual(
+            self.account.id,
+            result.provenance[
+                "current_automation_principal"
+            ]["account_id"],
+        )
 
     def test_revoked_scope_and_suspended_owner_fail_closed(self) -> None:
         context = UserAuthorizationContext(
@@ -151,6 +189,101 @@ class AutomationPrincipalTests(unittest.TestCase):
             suspended.provenance["status"],
         )
 
+    def test_service_account_resolution_uses_current_scope_ceiling(self) -> None:
+        account = Account(
+            id="service-account-backing",
+            email="service@example.invalid",
+            normalized_email="service@example.invalid",
+            display_name="Import worker",
+            auth_provider="service_account",
+        )
+        membership = User(
+            id="service-membership",
+            tenant_id=self.tenant.id,
+            account_id=account.id,
+            email=account.email,
+            display_name=account.display_name,
+            auth_provider="service_account",
+        )
+        service_account = ServiceAccount(
+            id="service-1",
+            tenant_id=self.tenant.id,
+            account_id=account.id,
+            membership_id=membership.id,
+            name="Import worker",
+            normalized_name="import worker",
+            scope_ceiling=[
+                "dataflow:pipeline:run",
+                "datasources:catalogue:read",
+                "system:settings:write",
+            ],
+            is_active=True,
+            revision=1,
+            settings={},
+        )
+        self.session.add_all(
+            (account, membership, service_account)
+        )
+        self.session.flush()
+        request = AutomationPrincipalRequest.service_account(
+            tenant_id=self.tenant.id,
+            service_account_id=service_account.id,
+            authorization_ref="dataflow-trigger:service",
+            grant_scopes=(
+                "dataflow:pipeline:run",
+                "datasources:catalogue:read",
+            ),
+        )
+
+        result = self.provider.resolve_automation_principal(
+            self.session,
+            request=request,
+        )
+
+        self.assertTrue(result.allowed)
+        self.assertEqual(
+            service_account.id,
+            result.principal.principal.service_account_id,
+        )
+        self.assertEqual(
+            frozenset(request.grant_scopes),
+            result.principal.scopes,
+        )
+        self.assertNotIn(
+            "system:settings:write",
+            result.principal.scopes,
+        )
+        self.assertEqual(
+            "service_account",
+            result.provenance["trigger_owner"]["kind"],
+        )
+
+        service_account.scope_ceiling = [
+            "dataflow:pipeline:run"
+        ]
+        self.session.flush()
+        reduced = self.provider.resolve_automation_principal(
+            self.session,
+            request=request,
+        )
+        self.assertFalse(reduced.allowed)
+        self.assertEqual(
+            ("datasources:catalogue:read",),
+            reduced.missing_scopes,
+        )
+
+        service_account.is_active = False
+        self.session.flush()
+        inactive = self.provider.resolve_automation_principal(
+            self.session,
+            request=request,
+        )
+        self.assertFalse(inactive.allowed)
+        self.assertEqual(
+            "inactive_or_inconsistent",
+            inactive.provenance["status"],
+        )
+
     def test_manifest_registers_automation_resolution(self) -> None:
         self.assertIn(
             CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,

tests/test_service_accounts.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db.base import AccessBase
+from govoplan_access.backend.db.models import (
+    Account,
+    ApiKey,
+    AuthSession,
+    User,
+)
+from govoplan_access.backend.service_accounts import (
+    ServiceAccountConflictError,
+    create_service_account,
+    retire_service_account,
+    update_service_account,
+)
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.tenancy.scope import (
+    Tenant,
+    create_scope_tables,
+    scope_registry,
+)
+
+
+class ServiceAccountTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        create_scope_tables(self.engine)
+        AccessBase.metadata.create_all(bind=self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.tenant = Tenant(
+            id="tenant-1",
+            slug="tenant-1",
+            name="Tenant 1",
+        )
+        self.account = Account(
+            id="account-1",
+            email="admin@example.test",
+            normalized_email="admin@example.test",
+        )
+        self.user = User(
+            id="user-1",
+            tenant_id=self.tenant.id,
+            account_id=self.account.id,
+            email=self.account.email,
+        )
+        self.session.add_all(
+            (self.tenant, self.account, self.user)
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        AccessBase.metadata.drop_all(bind=self.engine)
+        scope_registry.metadata.drop_all(bind=self.engine)
+        self.engine.dispose()
+
+    def _principal(
+        self,
+        scopes: frozenset[str] = frozenset({"tenant:*"}),
+    ) -> ApiPrincipal:
+        return ApiPrincipal(
+            principal=PrincipalRef(
+                account_id=self.account.id,
+                membership_id=self.user.id,
+                tenant_id=self.tenant.id,
+                scopes=scopes,
+            ),
+            account=self.account,
+            user=self.user,
+        )
+
+    def test_create_builds_a_non_login_identity_without_secrets(self) -> None:
+        item = create_service_account(
+            self.session,
+            tenant=self.tenant,
+            principal=self._principal(),
+            name=" Monthly import ",
+            description="Runs the governed monthly import.",
+            scope_ceiling=(
+                "dataflow:pipeline:run",
+                "datasources:catalogue:read",
+            ),
+        )
+
+        backing_account = self.session.get(
+            Account,
+            item.account_id,
+        )
+        membership = self.session.get(
+            User,
+            item.membership_id,
+        )
+        self.assertEqual("Monthly import", item.name)
+        self.assertEqual(
+            [
+                "dataflow:pipeline:run",
+                "datasources:catalogue:read",
+            ],
+            item.scope_ceiling,
+        )
+        self.assertEqual(
+            "service_account",
+            backing_account.auth_provider,
+        )
+        self.assertIsNone(backing_account.password_hash)
+        self.assertEqual(
+            "service_account",
+            membership.auth_provider,
+        )
+        self.assertIsNone(membership.password_hash)
+        self.assertEqual(
+            0,
+            self.session.query(ApiKey)
+            .filter(ApiKey.user_id == membership.id)
+            .count(),
+        )
+        self.assertEqual(
+            0,
+            self.session.query(AuthSession)
+            .filter(AuthSession.user_id == membership.id)
+            .count(),
+        )
+
+    def test_scope_escalation_and_stale_updates_fail_closed(self) -> None:
+        principal = self._principal(
+            frozenset({"dataflow:pipeline:run"})
+        )
+        with self.assertRaises(PermissionError):
+            create_service_account(
+                self.session,
+                tenant=self.tenant,
+                principal=principal,
+                name="Escalating worker",
+                description=None,
+                scope_ceiling=("system:settings:write",),
+            )
+
+        item = create_service_account(
+            self.session,
+            tenant=self.tenant,
+            principal=principal,
+            name="Bounded worker",
+            description=None,
+            scope_ceiling=("dataflow:pipeline:run",),
+        )
+        updated = update_service_account(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            principal=principal,
+            expected_revision=1,
+            changes={"description": "Updated"},
+        )
+        self.assertEqual(2, updated.revision)
+        with self.assertRaises(ServiceAccountConflictError):
+            update_service_account(
+                self.session,
+                tenant_id=self.tenant.id,
+                service_account_id=item.id,
+                principal=principal,
+                expected_revision=1,
+                changes={"description": "Stale"},
+            )
+
+    def test_retirement_revokes_the_backing_principal(self) -> None:
+        principal = self._principal()
+        item = create_service_account(
+            self.session,
+            tenant=self.tenant,
+            principal=principal,
+            name="Retired worker",
+            description=None,
+            scope_ceiling=("dataflow:pipeline:run",),
+        )
+
+        retired = retire_service_account(
+            self.session,
+            tenant_id=self.tenant.id,
+            service_account_id=item.id,
+            principal=principal,
+            expected_revision=1,
+        )
+
+        self.assertFalse(retired.is_active)
+        self.assertIsNotNone(retired.retired_at)
+        self.assertFalse(
+            self.session.get(Account, retired.account_id).is_active
+        )
+        self.assertFalse(
+            self.session.get(User, retired.membership_id).is_active
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()