feat(core): define sanctions screening gate contract

2026-07-30 01:54:58 +02:00
parent f3b388fe7e
commit ca8a8c5111
2 changed files with 323 additions and 1 deletions
@@ -3,13 +3,17 @@ from __future__ import annotations
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Protocol, runtime_checkable
+from typing import Literal, Protocol, runtime_checkable


 SANCTIONS_SNAPSHOT_CONTRACT_VERSION = "1"
+SANCTIONS_SCREENING_CONTRACT_VERSION = "1"
 CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS = (
    "connectors.sanctionsSnapshotProvider"
 )
+CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING = (
+    "riskCompliance.sanctionsScreeningProvider"
+)


@dataclass(frozen=True, slots=True)
@@ -90,6 +94,181 @@ class SanctionsSnapshotPayload:
            )


+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningSubject:
+    subject_type: Literal["person", "entity"]
+    primary_name: str | None = None
+    subject_ref: str | None = None
+    aliases: tuple[str, ...] = ()
+    identifiers: tuple[Mapping[str, str], ...] = ()
+    dates: tuple[str, ...] = ()
+    addresses: tuple[Mapping[str, str], ...] = ()
+
+    def __post_init__(self) -> None:
+        if self.subject_type not in {"person", "entity"}:
+            raise ValueError(
+                "Sanctions screening subjects must be a person or entity."
+            )
+        if not str(self.primary_name or "").strip() and not any(
+            str(value.get("value") or "").strip()
+            for value in self.identifiers
+        ):
+            raise ValueError(
+                "A sanctions screening subject needs a name or identifier."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningPolicy:
+    fuzzy_threshold: float = 0.88
+    max_snapshot_age_days: int = 7
+    max_candidates: int = 100
+    failure_policy: Literal["block", "review", "degraded"] = "block"
+
+    def __post_init__(self) -> None:
+        if not 0.8 <= self.fuzzy_threshold <= 1:
+            raise ValueError(
+                "Sanctions screening fuzzy threshold must be between 0.8 and 1."
+            )
+        if not 1 <= self.max_snapshot_age_days <= 365:
+            raise ValueError(
+                "Sanctions snapshot age must be between 1 and 365 days."
+            )
+        if not 1 <= self.max_candidates <= 100:
+            raise ValueError(
+                "Sanctions screening candidate limit must be between 1 and 100."
+            )
+        if self.failure_policy not in {"block", "review", "degraded"}:
+            raise ValueError(
+                "Sanctions screening failure policy is not supported."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningRequest:
+    list_snapshot_id: str
+    idempotency_key: str
+    subject: SanctionsScreeningSubject
+    policy: SanctionsScreeningPolicy = field(
+        default_factory=SanctionsScreeningPolicy
+    )
+    contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION:
+            raise ValueError(
+                "Unsupported sanctions screening contract version."
+            )
+        if not self.list_snapshot_id.strip():
+            raise ValueError(
+                "A sanctions list snapshot is required for screening."
+            )
+        if not self.idempotency_key.strip():
+            raise ValueError(
+                "A sanctions screening idempotency key is required."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningEvidence:
+    ref: str
+    run_id: str
+    outcome: str
+    candidate_count: int
+    list_snapshot_id: str
+    source_version: str
+    subject_fingerprint: str
+    matcher_version: str
+    normalization_version: str
+    policy_version: str
+    completed_at: datetime | None
+    contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION:
+            raise ValueError(
+                "Unsupported sanctions screening evidence version."
+            )
+        if self.ref != f"risk-screening:{self.run_id}":
+            raise ValueError(
+                "Sanctions screening evidence reference is invalid."
+            )
+        if self.candidate_count < 0:
+            raise ValueError(
+                "Sanctions screening candidate count cannot be negative."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningFreshnessRequest:
+    evidence_ref: str
+    current_subject: SanctionsScreeningSubject | None = None
+    expected_list_snapshot_id: str | None = None
+    policy: SanctionsScreeningPolicy = field(
+        default_factory=SanctionsScreeningPolicy
+    )
+    contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION:
+            raise ValueError(
+                "Unsupported sanctions screening freshness contract version."
+            )
+        if not self.evidence_ref.startswith("risk-screening:"):
+            raise ValueError(
+                "Sanctions screening evidence reference is invalid."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningFreshness:
+    evidence: SanctionsScreeningEvidence
+    fresh: bool
+    reasons: tuple[str, ...]
+    checked_at: datetime
+    current_list_snapshot_id: str | None
+    gate_decision: Literal["allow", "block", "review", "degraded"]
+    gate_reasons: tuple[str, ...]
+    contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION:
+            raise ValueError(
+                "Unsupported sanctions screening freshness version."
+            )
+        if self.fresh != (not self.reasons):
+            raise ValueError(
+                "Sanctions screening freshness reasons are inconsistent."
+            )
+        if self.gate_decision not in {
+            "allow",
+            "block",
+            "review",
+            "degraded",
+        }:
+            raise ValueError(
+                "Sanctions screening gate decision is invalid."
+            )
+
+
+@dataclass(frozen=True, slots=True)
+class SanctionsScreeningResult:
+    evidence: SanctionsScreeningEvidence
+    freshness: SanctionsScreeningFreshness
+    created: bool
+    contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION
+
+    def __post_init__(self) -> None:
+        if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION:
+            raise ValueError(
+                "Unsupported sanctions screening result version."
+            )
+        if self.evidence != self.freshness.evidence:
+            raise ValueError(
+                "Sanctions screening result evidence is inconsistent."
+            )
+
+
@runtime_checkable
 class SanctionsSnapshotProvider(Protocol):
    def list_snapshots(
@@ -120,6 +299,25 @@ class SanctionsSnapshotProvider(Protocol):
        ...


+@runtime_checkable
+class SanctionsScreeningProvider(Protocol):
+    def request_screening(
+        self,
+        session: object,
+        principal: object,
+        request: SanctionsScreeningRequest,
+    ) -> SanctionsScreeningResult:
+        ...
+
+    def check_freshness(
+        self,
+        session: object,
+        principal: object,
+        request: SanctionsScreeningFreshnessRequest,
+    ) -> SanctionsScreeningFreshness:
+        ...
+
+
 def sanctions_snapshot_provider(
    registry: object | None,
 ) -> SanctionsSnapshotProvider | None:
@@ -142,11 +340,44 @@ def sanctions_snapshot_provider(
    )


+def sanctions_screening_provider(
+    registry: object | None,
+) -> SanctionsScreeningProvider | None:
+    if (
+        registry is None
+        or not hasattr(registry, "has_capability")
+        or not hasattr(registry, "capability")
+        or not registry.has_capability(
+            CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING
+        )
+    ):
+        return None
+    provider = registry.capability(
+        CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING
+    )
+    return (
+        provider
+        if isinstance(provider, SanctionsScreeningProvider)
+        else None
+    )
+
+
 __all__ = [
    "CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS",
+    "CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING",
+    "SANCTIONS_SCREENING_CONTRACT_VERSION",
    "SANCTIONS_SNAPSHOT_CONTRACT_VERSION",
+    "SanctionsScreeningEvidence",
+    "SanctionsScreeningFreshness",
+    "SanctionsScreeningFreshnessRequest",
+    "SanctionsScreeningPolicy",
+    "SanctionsScreeningProvider",
+    "SanctionsScreeningRequest",
+    "SanctionsScreeningResult",
+    "SanctionsScreeningSubject",
    "SanctionsSnapshotPayload",
    "SanctionsSnapshotProvider",
    "SanctionsSnapshotReference",
+    "sanctions_screening_provider",
    "sanctions_snapshot_provider",
 ]