Prepare custody observation adapter pending authenticated owner delivery

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29

src/user_engine/custody.py

@@ -0,0 +1,119 @@
+"""Proposed NK-ADR-0017 observation seam; deliberately absent from runtime.
+
+Reader and baseline implementations are deployment-owner responsibilities.
+No claims, account roles, fixture identities or local cache can create a grant.
+"""
+from __future__ import annotations
+
+from copy import deepcopy
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Mapping, Protocol
+
+
+@dataclass(frozen=True)
+class CustodyRead:
+    source_id: str
+    document: Mapping[str, Any]
+    read_at: datetime
+
+
+@dataclass(frozen=True)
+class CustodyBaseline:
+    source_id: str
+    document: Mapping[str, Any]
+
+
+class CustodyReader(Protocol):
+    def read(self) -> CustodyRead:
+        """Authenticated fresh owner read, bounded by timeout and response size.
+
+        source_id is trusted transport configuration, never response metadata.
+        Return actual successful owner read time; raise on failure or cache hits.
+        """
+        ...
+
+
+class CustodyBaselineStore(Protocol):
+    def load(self) -> CustodyBaseline | None: ...
+
+    def compare_and_store(
+        self, previous: CustodyBaseline | None, current: CustodyBaseline,
+    ) -> bool:
+        """Atomically persist only if previous still matches; survive restart.
+
+        Return false on conflict or write failure. Never reset on source changes.
+        """
+        ...
+
+
+class CustodyPolicy(Protocol):
+    """NetKingdom's version-pinned validator and observe_root_grant functions."""
+    def validate(self, record: Any) -> list[str]: ...
+    def successor_errors(self, previous: dict, current: dict) -> list[str]: ...
+    def observe_root_grant(
+        self, record: dict, issuer: str, subject: str, tenant: str,
+        read_at: datetime, now: datetime, previous_record: dict | None = None,
+    ) -> dict | None: ...
+
+
+class CustodyObservationAdapter:
+    def __init__(
+        self, *, reader: CustodyReader, baselines: CustodyBaselineStore,
+        policy: CustodyPolicy, source_id: str, record_id: str,
+        allow_initial_baseline: bool = False,
+    ):
+        if not source_id or not record_id:
+            raise ValueError("trusted source and record bindings required")
+        self.reader, self.baselines, self.policy = reader, baselines, policy
+        self.source_id, self.record_id = source_id, record_id
+        self.allow_initial_baseline = allow_initial_baseline
+
+    def observe(
+        self, *, issuer: str, subject: str, tenant: str, now: datetime,
+    ) -> dict | None:
+        """Always re-read; failures never reuse or refresh an old observation.
+
+        Diagnostics are suppressed because transport/validator errors may contain
+        credentials or rejected data. This supplies a fact, never authorization.
+        """
+        if tenant != "platform" or now.utcoffset() is None:
+            return None
+        try:
+            previous = self.baselines.load()
+            if previous is None and not self.allow_initial_baseline:
+                return None
+            if previous is not None and (
+                previous.source_id != self.source_id
+                or previous.document.get("record_id") != self.record_id
+                or self.policy.validate(previous.document)
+            ):
+                return None
+            read = self.reader.read()
+            document = deepcopy(dict(read.document))
+            if (read.source_id != self.source_id
+                or document.get("record_id") != self.record_id
+                or document.get("owner_system") != "net-kingdom-custody"
+                or read.read_at.utcoffset() is None
+                or read.read_at > now
+                or self.policy.validate(document)):
+                return None
+            old = deepcopy(dict(previous.document)) if previous else None
+            if old is not None and self.policy.successor_errors(old, document):
+                return None
+            # Reject stale/future documents even when they contain no grant.
+            updated = datetime.fromisoformat(document["updated_at"].replace("Z", "+00:00"))
+            age = (now - read.read_at).total_seconds()
+            if updated > read.read_at or not 0 <= age <= document["freshness"]["max_cache_seconds"]:
+                return None
+            result = self.policy.observe_root_grant(
+                document, issuer, subject, tenant, read.read_at, now,
+                previous_record=old,
+            )
+            # Keep revocation/empty revisions too; absence does not reset history.
+            current = CustodyBaseline(self.source_id, deepcopy(document))
+            if not self.baselines.compare_and_store(previous, current):
+                return None
+            return deepcopy(result)
+        except Exception:
+            return None