Expose provider-neutral identity reconciliation

src/user_engine/adapters/provisioning.py

@@ -3,20 +3,29 @@
 from __future__ import annotations
 
 import json
+import time
 from typing import Any
-from urllib.error import HTTPError
+from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 
-from user_engine.ports import ProvisioningRequest, ProvisioningResult
+from user_engine.ports import IdentityDriftResult, ProvisioningRequest, ProvisioningResult
 
 
 class HTTPIdentityProvisioningAdapter:
-    def __init__(self, *, base_url: str, bearer_token: str, timeout: float = 10) -> None:
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        bearer_token: str,
+        timeout: float = 10,
+        retry_delay: float = 0.1,
+    ) -> None:
         self.base_url = base_url.rstrip("/")
         self.bearer_token = bearer_token.strip()
         if not self.bearer_token:
             raise ValueError("bearer token must not be empty")
         self.timeout = timeout
+        self.retry_delay = retry_delay
 
     def provision(self, request: ProvisioningRequest) -> ProvisioningResult:
         return self._post("/v1/identities/provision", {
@@ -38,6 +47,24 @@ class HTTPIdentityProvisioningAdapter:
     def deprovision(self, *, external_subject: str, idempotency_key: str, correlation_id: str) -> ProvisioningResult:
         return self._lifecycle("deprovision", external_subject, idempotency_key, correlation_id)
 
+    def drift(
+        self,
+        request: ProvisioningRequest,
+        *,
+        external_subject: str,
+        desired_status: str = "active",
+    ) -> IdentityDriftResult:
+        return self._desired("drift", request, external_subject, desired_status)
+
+    def reconcile(
+        self,
+        request: ProvisioningRequest,
+        *,
+        external_subject: str,
+        desired_status: str = "active",
+    ) -> IdentityDriftResult:
+        return self._desired("reconcile", request, external_subject, desired_status)
+
     def _lifecycle(self, action: str, subject: str, key: str, correlation_id: str) -> ProvisioningResult:
         return self._post(f"/v1/identities/{action}", {
             "external_subject": subject,
@@ -45,22 +72,34 @@ class HTTPIdentityProvisioningAdapter:
             "correlation_id": correlation_id,
         })
 
-    def _post(self, path: str, payload: dict[str, Any]) -> ProvisioningResult:
-        request = Request(
-            self.base_url + path,
-            data=json.dumps(payload).encode(),
-            headers={
-                "Authorization": f"Bearer {self.bearer_token}",
-                "Content-Type": "application/json",
-            },
-            method="POST",
+    def _desired(
+        self,
+        action: str,
+        request: ProvisioningRequest,
+        external_subject: str,
+        desired_status: str,
+    ) -> IdentityDriftResult:
+        result = self._request(f"/v1/identities/{action}", {
+            "external_subject": external_subject,
+            "user_id": request.user_id,
+            "tenant": request.tenant,
+            "primary_email": request.primary_email,
+            "display_name": request.display_name,
+            "idempotency_key": request.idempotency_key,
+            "correlation_id": request.correlation_id,
+            "roles": request.roles,
+            "desired_status": desired_status,
+        })
+        return IdentityDriftResult(
+            provider=str(result["provider"]),
+            external_subject=str(result["external_subject"]),
+            status=str(result["status"]),
+            drift=tuple(str(item) for item in result.get("drift", ())),
+            changed=tuple(str(item) for item in result.get("changed", ())),
         )
-        try:
-            with urlopen(request, timeout=self.timeout) as response:
-                result = json.loads(response.read())
-        except HTTPError as exc:
-            message = exc.read(4096).decode("utf-8", "replace")
-            raise RuntimeError(f"identity provisioning failed ({exc.code}): {message}") from exc
+
+    def _post(self, path: str, payload: dict[str, Any]) -> ProvisioningResult:
+        result = self._request(path, payload)
         return ProvisioningResult(
             provider=str(result["provider"]),
             external_subject=str(result["external_subject"]),
@@ -72,3 +111,30 @@ class HTTPIdentityProvisioningAdapter:
                 else None
             ),
         )
+
+    def _request(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
+        request = Request(
+            self.base_url + path,
+            data=json.dumps(payload).encode(),
+            headers={
+                "Authorization": f"Bearer {self.bearer_token}",
+                "Content-Type": "application/json",
+            },
+            method="POST",
+        )
+        for attempt in range(2):
+            try:
+                with urlopen(request, timeout=self.timeout) as response:
+                    result = json.loads(response.read())
+                break
+            except HTTPError as exc:
+                if exc.code not in {502, 503, 504} or attempt:
+                    message = exc.read(4096).decode("utf-8", "replace")
+                    raise RuntimeError(
+                        f"identity provisioning failed ({exc.code}): {message}"
+                    ) from exc
+            except URLError as exc:
+                if attempt:
+                    raise RuntimeError("identity provisioning dependency unavailable") from exc
+            time.sleep(self.retry_delay)
+        return dict(result)

src/user_engine/ports.py

@@ -66,6 +66,15 @@ class ProvisioningResult:
     password_setup_url: str | None = None
 
 
+@dataclass(frozen=True)
+class IdentityDriftResult:
+    provider: str
+    external_subject: str
+    status: str
+    drift: tuple[str, ...] = ()
+    changed: tuple[str, ...] = ()
+
+
 class IdentityProvisioningPort(Protocol):
     """Lifecycle seam owned by NetKingdom adapters, not the user domain."""
 
@@ -87,6 +96,24 @@ class IdentityProvisioningPort(Protocol):
     ) -> ProvisioningResult:
         """Remove or tombstone an identity according to provider policy."""
 
+    def drift(
+        self,
+        request: ProvisioningRequest,
+        *,
+        external_subject: str,
+        desired_status: str = "active",
+    ) -> IdentityDriftResult:
+        """Inspect provider state without changing it or exposing credentials."""
+
+    def reconcile(
+        self,
+        request: ProvisioningRequest,
+        *,
+        external_subject: str,
+        desired_status: str = "active",
+    ) -> IdentityDriftResult:
+        """Converge managed provider state toward the requested lifecycle."""
+
 
 class UserEngineStore(Protocol):
     """Durable persistence boundary for user-engine service behavior.