user-engine/src/user_engine/ports.py

"""Implementation ports for user-engine adapters.

The domain layer should depend on these protocols. Concrete implementations
can be local test adapters, HTTP clients, database-backed stores, or platform
adapters without changing domain code.
"""

from __future__ import annotations

from typing import Any, Iterable, Mapping, Protocol

from user_engine.domain import (
    Actor,
    ApplicationBinding,
    AuditRecord,
    AuthorizationDecision,
    AuthorizationRequest,
    CanonEntityReference,
    Membership,
    OutboxEvent,
)


class IdentityClaimsAdapter(Protocol):
    """Normalize verified identity claims into a user-engine actor."""

    def normalize(self, claims: Mapping[str, Any]) -> Actor:
        """Return a normalized actor from already-verified claims."""

    def identity_key(self, actor: Actor) -> tuple[str, str]:
        """Return the stable external identity link key."""


class AuthorizationCheckPort(Protocol):
    """Ask whether an actor may perform an action."""

    def check(self, request: AuthorizationRequest) -> AuthorizationDecision:
        """Return the authorization decision for one request."""

    def batch_check(
        self, requests: Iterable[AuthorizationRequest]
    ) -> tuple[AuthorizationDecision, ...]:
        """Return decisions in request order."""


class ApplicationBindingStore(Protocol):
    """Store links between user-engine applications and external systems."""

    def get(self, application_id: str) -> ApplicationBinding | None:
        """Return a binding by user-engine application id."""

    def save(self, binding: ApplicationBinding) -> None:
        """Create or replace an application binding."""


class MembershipFactExporter(Protocol):
    """Export membership facts as read models for authorization systems."""

    def export(self, memberships: Iterable[Membership]) -> Mapping[str, Any]:
        """Return an adapter-neutral membership fact manifest."""


class EventOutbox(Protocol):
    """Persist and publish durable domain events."""

    def append(self, event: OutboxEvent) -> None:
        """Append an event in the same unit of work as its mutation."""

    def pending(self) -> tuple[OutboxEvent, ...]:
        """Return events waiting for delivery."""


class AuditWriter(Protocol):
    """Persist local audit records and support platform audit export."""

    def record(self, audit_record: AuditRecord) -> None:
        """Persist an audit record."""


class EvidenceReferenceExporter(Protocol):
    """Export audit/review material as identity-canon evidence references."""

    def export(
        self, audit_records: Iterable[AuditRecord]
    ) -> tuple[CanonEntityReference, ...]:
        """Return evidence references without owning the platform audit sink."""


class PolicyControlReferenceResolver(Protocol):
    """Resolve policy/control references for identity-domain traces."""

    def references_for(
        self, request: AuthorizationRequest, decision: AuthorizationDecision
    ) -> Mapping[str, CanonEntityReference]:
        """Return policy, control, review, or exception references when known."""


class LifecycleTaskSink(Protocol):
    """Handoff identity-domain gaps or lifecycle work to a task system."""

    def create_or_link(
        self,
        *,
        summary: str,
        subject: CanonEntityReference,
        evidence: Iterable[CanonEntityReference] = (),
    ) -> CanonEntityReference:
        """Return the task reference created or linked by the downstream system."""


class SecretProvider(Protocol):
    """Load runtime secret material from the active environment."""

    def get(self, name: str) -> str:
        """Return a secret value by logical name."""