Implement durable store contract and registration roadmap

src/user_engine/adapters/local.py

@@ -2,8 +2,10 @@
 
 from __future__ import annotations
 
+import copy
+from contextlib import contextmanager
 from dataclasses import dataclass, field
-from typing import Iterable
+from typing import Iterable, Iterator, Mapping, cast
 
 from user_engine.domain import (
     Account,
@@ -51,6 +53,10 @@ class InMemoryUserEngineStore:
     ] = field(default_factory=dict)
     audit_records: list[AuditRecord] = field(default_factory=list)
     outbox_events: list[OutboxEvent] = field(default_factory=list)
+    _transaction_depth: int = field(default=0, init=False, repr=False)
+    _transaction_snapshot: Mapping[str, object] | None = field(
+        default=None, init=False, repr=False
+    )
 
     def migrate(self) -> None:
         """Apply the standalone schema migration manifest."""
@@ -60,15 +66,42 @@ class InMemoryUserEngineStore:
     def ready(self) -> bool:
         return self.schema_version == SCHEMA_VERSION
 
+    @contextmanager
+    def transaction(self) -> Iterator[None]:
+        """Provide atomic in-memory mutation semantics for conformance tests."""
+        if self._transaction_depth == 0:
+            self._transaction_snapshot = self._snapshot()
+        self._transaction_depth += 1
+        try:
+            yield
+        except Exception:
+            if self._transaction_depth == 1 and self._transaction_snapshot is not None:
+                self._restore(self._transaction_snapshot)
+            raise
+        finally:
+            self._transaction_depth -= 1
+            if self._transaction_depth == 0:
+                self._transaction_snapshot = None
+
     def save_user(self, user: User) -> None:
         self.users[user.user_id] = user
 
+    def user(self, user_id: str) -> User | None:
+        return self.users.get(user_id)
+
     def save_account(self, account: Account) -> None:
         self.accounts[account.user_id] = account
 
     def save_identity(self, identity: ExternalIdentity) -> None:
         self.identities[identity.identity_key] = identity
 
+    def identities_for_user(self, user_id: str) -> tuple[ExternalIdentity, ...]:
+        return tuple(
+            identity
+            for identity in self.identities.values()
+            if identity.user_id == user_id
+        )
+
     def save_tenant_account(self, account: TenantAccount) -> None:
         self.tenant_accounts[(account.tenant, account.user_id)] = account
 
@@ -78,12 +111,24 @@ class InMemoryUserEngineStore:
     def save_application(self, application: Application) -> None:
         self.applications[application.application_id] = application
 
+    def application(self, application_id: str) -> Application | None:
+        return self.applications.get(application_id)
+
     def save_binding(self, binding: ApplicationBinding) -> None:
         self.bindings[binding.application_id] = binding
 
+    def binding(self, application_id: str) -> ApplicationBinding | None:
+        return self.bindings.get(application_id)
+
     def save_catalog(self, catalog: Catalog) -> None:
         self.catalogs[catalog.catalog_id] = catalog
 
+    def catalog(self, catalog_id: str) -> Catalog | None:
+        return self.catalogs.get(catalog_id)
+
+    def all_catalogs(self) -> tuple[Catalog, ...]:
+        return tuple(self.catalogs.values())
+
     def save_family_invitation(self, invitation: FamilyInvitation) -> None:
         self.family_invitations[invitation.invitation_id] = invitation
 
@@ -136,9 +181,67 @@ class InMemoryUserEngineStore:
     def append_audit(self, record: AuditRecord) -> None:
         self.audit_records.append(record)
 
+    def audit_log(self) -> tuple[AuditRecord, ...]:
+        return tuple(self.audit_records)
+
     def append_outbox(self, event: OutboxEvent) -> None:
         self.outbox_events.append(event)
 
+    def pending_outbox(self) -> tuple[OutboxEvent, ...]:
+        return tuple(self.outbox_events)
+
+    def record_counts(self) -> Mapping[str, int]:
+        return {
+            "users": len(self.users),
+            "accounts": len(self.accounts),
+            "tenant_accounts": len(self.tenant_accounts),
+            "memberships": len(self.memberships),
+            "applications": len(self.applications),
+            "catalogs": len(self.catalogs),
+            "family_invitations": len(self.family_invitations),
+            "profile_values": len(self.profile_values),
+            "audit_records": len(self.audit_records),
+            "pending_outbox_events": len(self.outbox_events),
+        }
+
+    def _snapshot(self) -> Mapping[str, object]:
+        return {
+            "users": copy.deepcopy(self.users),
+            "accounts": copy.deepcopy(self.accounts),
+            "identities": copy.deepcopy(self.identities),
+            "tenant_accounts": copy.deepcopy(self.tenant_accounts),
+            "memberships": copy.deepcopy(self.memberships),
+            "applications": copy.deepcopy(self.applications),
+            "bindings": copy.deepcopy(self.bindings),
+            "catalogs": copy.deepcopy(self.catalogs),
+            "family_invitations": copy.deepcopy(self.family_invitations),
+            "profile_values": copy.deepcopy(self.profile_values),
+            "audit_records": copy.deepcopy(self.audit_records),
+            "outbox_events": copy.deepcopy(self.outbox_events),
+        }
+
+    def _restore(self, snapshot: Mapping[str, object]) -> None:
+        snapshot_audit_records = cast(list[AuditRecord], snapshot["audit_records"])
+        denied_audit_records = [
+            record
+            for record in self.audit_records[len(snapshot_audit_records) :]
+            if record.summary == "authorization denied"
+        ]
+        self.users = snapshot["users"]  # type: ignore[assignment]
+        self.accounts = snapshot["accounts"]  # type: ignore[assignment]
+        self.identities = snapshot["identities"]  # type: ignore[assignment]
+        self.tenant_accounts = snapshot["tenant_accounts"]  # type: ignore[assignment]
+        self.memberships = snapshot["memberships"]  # type: ignore[assignment]
+        self.applications = snapshot["applications"]  # type: ignore[assignment]
+        self.bindings = snapshot["bindings"]  # type: ignore[assignment]
+        self.catalogs = snapshot["catalogs"]  # type: ignore[assignment]
+        self.family_invitations = snapshot[
+            "family_invitations"
+        ]  # type: ignore[assignment]
+        self.profile_values = snapshot["profile_values"]  # type: ignore[assignment]
+        self.audit_records = [*snapshot_audit_records, *denied_audit_records]
+        self.outbox_events = snapshot["outbox_events"]  # type: ignore[assignment]
+
 
 class LocalAuthorizationCheckPort:
     """Deterministic local authorization adapter.

src/user_engine/ports.py

@@ -7,20 +7,141 @@ adapters without changing domain code.
 
 from __future__ import annotations
 
+from contextlib import AbstractContextManager
 from typing import Any, Iterable, Mapping, Protocol
 
 from user_engine.domain import (
+    Account,
     Actor,
+    Application,
     ApplicationBinding,
     AuditRecord,
     AuthorizationDecision,
     AuthorizationRequest,
     CanonEntityReference,
+    Catalog,
+    ExternalIdentity,
+    FamilyInvitation,
     Membership,
     OutboxEvent,
+    ProfileValue,
+    TenantAccount,
+    User,
 )
 
 
+class UserEngineStore(Protocol):
+    """Durable persistence boundary for user-engine service behavior.
+
+    Implementations may be in-memory, Postgres-backed, or platform-provided,
+    but must preserve the same logical keys, readiness contract, and atomic
+    mutation semantics exposed here.
+    """
+
+    schema_version: str | None
+
+    @property
+    def ready(self) -> bool:
+        """Return whether the store is schema-compatible for service use."""
+
+    def migrate(self) -> None:
+        """Apply or verify user-engine-owned schema migrations."""
+
+    def transaction(self) -> AbstractContextManager[None]:
+        """Return a context manager for one atomic mutation unit."""
+
+    def save_user(self, user: User) -> None:
+        """Create or replace a user record."""
+
+    def user(self, user_id: str) -> User | None:
+        """Return a user by id."""
+
+    def save_account(self, account: Account) -> None:
+        """Create or replace a primary account record."""
+
+    def user_account(self, user_id: str) -> Account | None:
+        """Return the primary account for a user."""
+
+    def save_identity(self, identity: ExternalIdentity) -> None:
+        """Create or replace an external identity link."""
+
+    def find_identity(self, issuer: str, subject: str) -> ExternalIdentity | None:
+        """Return an external identity by issuer and subject."""
+
+    def identities_for_user(self, user_id: str) -> tuple[ExternalIdentity, ...]:
+        """Return all external identities linked to a user."""
+
+    def save_tenant_account(self, account: TenantAccount) -> None:
+        """Create or replace a tenant-scoped account record."""
+
+    def tenant_account(self, tenant: str, user_id: str) -> TenantAccount | None:
+        """Return a tenant-scoped account record."""
+
+    def save_membership(self, membership: Membership) -> None:
+        """Create or replace a membership fact."""
+
+    def memberships_for_user(
+        self, user_id: str, *, tenant: str | None = None
+    ) -> tuple[Membership, ...]:
+        """Return memberships for a user, optionally scoped to a tenant."""
+
+    def memberships_for_tenant(self, tenant: str) -> tuple[Membership, ...]:
+        """Return memberships scoped to a tenant."""
+
+    def save_application(self, application: Application) -> None:
+        """Create or replace an application registration."""
+
+    def application(self, application_id: str) -> Application | None:
+        """Return an application by id."""
+
+    def save_binding(self, binding: ApplicationBinding) -> None:
+        """Create or replace an application binding."""
+
+    def binding(self, application_id: str) -> ApplicationBinding | None:
+        """Return an application binding by application id."""
+
+    def save_catalog(self, catalog: Catalog) -> None:
+        """Create or replace a catalog."""
+
+    def catalog(self, catalog_id: str) -> Catalog | None:
+        """Return a catalog by id."""
+
+    def all_catalogs(self) -> tuple[Catalog, ...]:
+        """Return all catalogs."""
+
+    def save_family_invitation(self, invitation: FamilyInvitation) -> None:
+        """Create or replace a family invitation."""
+
+    def family_invitation(self, invitation_id: str) -> FamilyInvitation | None:
+        """Return a family invitation by id."""
+
+    def family_invitations_for_user(
+        self, user_id: str
+    ) -> tuple[FamilyInvitation, ...]:
+        """Return family invitations for a user."""
+
+    def save_profile_value(self, value: ProfileValue) -> None:
+        """Create or replace a profile value."""
+
+    def values_for_user(self, user_id: str) -> tuple[ProfileValue, ...]:
+        """Return profile values for a user."""
+
+    def append_audit(self, record: AuditRecord) -> None:
+        """Append a local audit record."""
+
+    def audit_log(self) -> tuple[AuditRecord, ...]:
+        """Return local audit records in write order."""
+
+    def append_outbox(self, event: OutboxEvent) -> None:
+        """Append an outbox event."""
+
+    def pending_outbox(self) -> tuple[OutboxEvent, ...]:
+        """Return pending outbox events in write order."""
+
+    def record_counts(self) -> Mapping[str, int]:
+        """Return adapter-neutral record counts for diagnostics."""
+
+
 class IdentityClaimsAdapter(Protocol):
     """Normalize verified identity claims into a user-engine actor."""