from __future__ import annotations from pathlib import Path from govoplan_core.core.access import ( CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) from govoplan_core.core.module_guards import ( drop_table_retirement_provider, persistent_table_uninstall_guard, ) from govoplan_core.core.modules import ( DocumentationLink, DocumentationTopic, FrontendModule, FrontendRoute, MigrationSpec, ModuleInterfaceProvider, ModuleInterfaceRequirement, ModuleManifest, NavItem, PermissionDefinition, RoleTemplate, ) from govoplan_core.core.sanctions import ( CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING, ) from govoplan_core.core.search import SearchSourceProviderRegistration from govoplan_core.core.provider_governance import ( ModuleArchitectureDeclaration, ModuleArchitectureDocumentation, ModuleMaturityEvidence, ) from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_risk_compliance.backend.db.models import ( RiskAssuranceEdge, RiskAssuranceNode, RiskSanctionsAddress, RiskSanctionsAlias, RiskSanctionsDate, RiskSanctionsEntry, RiskSanctionsIdentifier, RiskSanctionsListSnapshot, RiskScreeningCandidate, RiskScreeningDisposition, RiskScreeningException, RiskScreeningRun, RiskScreeningSubjectSnapshot, ) from govoplan_risk_compliance.backend.permissions import ( ADMIN_SCOPE, READ_SCOPE, SANCTIONS_ADMIN_SCOPE, SANCTIONS_READ_SCOPE, SANCTIONS_REVIEW_SCOPE, SANCTIONS_SCREEN_SCOPE, WRITE_SCOPE, ) MODULE_ID = "risk_compliance" MODULE_NAME = "Risk Compliance" MODULE_VERSION = "0.1.17" OPTIONAL_DEPENDENCIES = ( "audit", "policy", "records", "inspections", "files", "tasks", "notifications", "connectors", ) _PERSISTENT_MODELS = ( RiskAssuranceEdge, RiskAssuranceNode, RiskScreeningException, RiskScreeningDisposition, RiskScreeningCandidate, RiskScreeningRun, RiskScreeningSubjectSnapshot, RiskSanctionsAddress, RiskSanctionsDate, RiskSanctionsIdentifier, RiskSanctionsAlias, RiskSanctionsEntry, RiskSanctionsListSnapshot, ) ARCHITECTURE = ModuleArchitectureDeclaration( layer="governance_accountability", kind="governance", maturity="vertical_slice", evidence=( ModuleMaturityEvidence( kind="test", reference="tests/test_assurance_graph.py", summary=( "Exercises tenant-safe immutable graph revisions, bounded traversal, " "sanctions projection, synthetic controls, ACL, and search." ), ), ModuleMaturityEvidence( kind="test", reference="tests/test_sanctions_screening.py", summary=( "Exercises immutable sanctions evidence, deterministic matching, " "review, exceptions, and freshness gates." ), ), ModuleMaturityEvidence( kind="migration", reference="tests/test_migrations.py", summary=( "Exercises the persistent sanctions screening and assurance " "graph schemas." ), ), ModuleMaturityEvidence( kind="documentation", reference="docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md", summary="Defines assurance ownership and integration boundaries.", ), ), known_limits=( "The assurance graph provides generic governance primitives; domain modules still own corrective execution.", "Cross-tenant aggregate assurance is intentionally not exposed by the tenant API.", ), supported_authority_modes=( "native_authoritative", "external_mirror", "governance_overlay", "linked_reference", ), owned_concepts=( "risk and control evaluation", "sanctions screening runs", "candidate review and dispositions", "compliance findings and assurance review", ), non_owned_concepts=( "external source transport and credentials", "immutable audit event storage", "policy rule evaluation", "governed domain objects and corrective execution", ), documentation=ModuleArchitectureDocumentation( migration=("tests/test_migrations.py",), upgrade=("docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md",), recovery=("docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md",), security=("docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md",), operations=("docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md",), ), ) def _permission( scope: str, label: str, description: str, ) -> PermissionDefinition: module_id, resource, action = scope.split(":", 2) return PermissionDefinition( scope=scope, label=label, description=description, category="Risk Compliance", level="tenant", module_id=module_id, resource=resource, action=action, ) PERMISSIONS = ( _permission( READ_SCOPE, "View risk compliance workspace", "Read risk compliance records, configuration, and workflow context.", ), _permission( WRITE_SCOPE, "Manage risk compliance workspace", "Create and update risk compliance records and workflow state.", ), _permission( ADMIN_SCOPE, "Administer risk compliance workspace", "Configure risk compliance policies, templates, and tenant administration.", ), _permission( SANCTIONS_READ_SCOPE, "View sanctions screening evidence", "Read list snapshots, screening runs, and candidate evidence.", ), _permission( SANCTIONS_SCREEN_SCOPE, "Run sanctions screening", "Submit subjects for deterministic screening against an immutable list.", ), _permission( SANCTIONS_REVIEW_SCOPE, "Review sanctions candidates", "Record evidence-backed candidate dispositions.", ), _permission( SANCTIONS_ADMIN_SCOPE, "Administer sanctions screening", "Import source snapshots, configure policy, and authorize review overrides.", ), ) ROLE_TEMPLATES = ( RoleTemplate( slug="risk_compliance_manager", name="Risk Compliance manager", description=("Manage compliance workflows and administer sanctions screening."), permissions=( READ_SCOPE, WRITE_SCOPE, SANCTIONS_READ_SCOPE, SANCTIONS_SCREEN_SCOPE, SANCTIONS_REVIEW_SCOPE, SANCTIONS_ADMIN_SCOPE, ), ), RoleTemplate( slug="risk_compliance_reviewer", name="Risk Compliance reviewer", description=("Run screenings and independently review potential matches."), permissions=( READ_SCOPE, SANCTIONS_READ_SCOPE, SANCTIONS_SCREEN_SCOPE, SANCTIONS_REVIEW_SCOPE, ), ), RoleTemplate( slug="risk_compliance_viewer", name="Risk Compliance viewer", description="Read risk compliance records and screening evidence.", permissions=(READ_SCOPE, SANCTIONS_READ_SCOPE), ), ) def _route_factory(_context): from govoplan_risk_compliance.backend.router import router return router def _sanctions_screening_provider(_context): from govoplan_risk_compliance.backend.capabilities import ( RiskComplianceSanctionsScreeningProvider, ) return RiskComplianceSanctionsScreeningProvider() def _assurance_search_source(context): from govoplan_risk_compliance.backend.search_source import ( create_risk_assurance_search_source, ) return create_risk_assurance_search_source(context) def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "risk_sanctions_list_snapshots": ( session.query(RiskSanctionsListSnapshot) .filter(RiskSanctionsListSnapshot.tenant_id == tenant_id) .count() ), "risk_screening_runs": ( session.query(RiskScreeningRun) .filter(RiskScreeningRun.tenant_id == tenant_id) .count() ), "risk_pending_screening_candidates": ( session.query(RiskScreeningCandidate) .filter( RiskScreeningCandidate.tenant_id == tenant_id, RiskScreeningCandidate.review_status.in_( ("pending", "exception_review") ), ) .count() ), "risk_assurance_nodes": ( session.query(RiskAssuranceNode) .filter( RiskAssuranceNode.tenant_id == tenant_id, RiskAssuranceNode.superseded_at.is_(None), ) .count() ), } DOCUMENTATION = ( DocumentationTopic( id=f"{MODULE_ID}.module-boundary", title=f"{MODULE_NAME} module boundary", summary=( "Risk and compliance workflows own legal evaluation, immutable " "screening evidence, review, and dispositions." ), body=( "Connectors may acquire source evidence, but Risk Compliance " "owns immutable normalized sanctions lists, version-pinned " "screening, candidate review, and legal dispositions. Fuzzy " "matching only creates candidates and never confirms a match. " "The broader module direction links obligations, governed object " "references, risks, controls, evidence, findings, corrective " "measures, and effectiveness reviews without copying the governed " "domain object or replacing Policy and Audit." ), layer="available", documentation_types=("admin", "user"), audience=( "operator", "module_admin", "compliance_reviewer", ), order=100, related_modules=OPTIONAL_DEPENDENCIES, links=( DocumentationLink( label="Repository domain boundary", href=( "govoplan-risk-compliance/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md" ), kind="repository", ), DocumentationLink( label="Interface pattern migration", href=( "govoplan-risk-compliance/docs/INTERFACE_PATTERN_MIGRATION.md" ), kind="repository", ), ), metadata={ "domain_objects": [ "sanctions list snapshots", "screening runs", "candidate evidence", "review dispositions", "time-bounded exceptions", ], "privacy": ( "Queue and audit summaries contain stable references and " "minimal subject data." ), "assurance_domain_model": [ "obligation", "governed object reference", "risk", "control", "evidence", "finding", "corrective measure", "effectiveness review", ], "assurance_graph": ( "Every node and edge is effective-dated, revisioned, tenant-scoped, " "and linked through opaque governed-object references." ), "help_contexts": [ "risk_compliance.workspace", "risk_compliance.sanctions.sources", "risk_compliance.action.import-snapshot", "risk_compliance.sanctions.screening", "risk_compliance.action.run-screening", "risk_compliance.sanctions.review", "risk_compliance.review.disposition", "risk_compliance.assurance.graph", "risk_compliance.assurance.editor", "risk_compliance.action.connect-assurance", "risk_compliance.state.source-unavailable", "risk_compliance.state.read-only", ], "consequence_classes": { "import_snapshot": "copy connector evidence into an immutable normalized sanctions-list snapshot", "run_screening": "create immutable version-pinned screening evidence from the minimum submitted subject data", "record_disposition": "append an evidence-backed legal review disposition that is not edited in place", "record_exception": "append a time-bounded subject-and-entry exception with explicit expiry", "revise_assurance_object": "append a new effective-dated revision while preserving prior evidence", "connect_assurance_objects": "append a governed typed relationship between assurance objects", }, }, ), ) manifest = ModuleManifest( id=MODULE_ID, name=MODULE_NAME, version=MODULE_VERSION, dependencies=("access",), optional_dependencies=OPTIONAL_DEPENDENCIES, required_capabilities=( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, ), provides_interfaces=( ModuleInterfaceProvider( name="risk_compliance.sanctions_screening", version="1.0.0", ), ), requires_interfaces=( ModuleInterfaceRequirement( name="connectors.sanctions_snapshots", version_min="1.0.0", version_max_exclusive="2.0.0", optional=True, ), ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_route_factory, capability_factories={ CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING: (_sanctions_screening_provider), }, search_sources=( SearchSourceProviderRegistration( id="risk_compliance.assurance", factory=_assurance_search_source, ), ), frontend=FrontendModule( module_id=MODULE_ID, package_name="@govoplan/risk-compliance-webui", routes=( FrontendRoute( path="/risk-compliance", component="RiskCompliancePage", required_any=(READ_SCOPE, SANCTIONS_READ_SCOPE), order=115, surface_id="risk_compliance.workspace", ), ), nav_items=( NavItem( path="/risk-compliance", label="Risk Compliance", icon="shield-check", required_any=(READ_SCOPE, SANCTIONS_READ_SCOPE), order=115, surface_id="risk_compliance.navigation", ), ), view_surfaces=( ViewSurface( id="risk_compliance.sanctions.sources", module_id=MODULE_ID, kind="section", label="Sanctions source snapshots", parent_id="risk_compliance.workspace", order=20, ), ViewSurface( id="risk_compliance.sanctions.screening", module_id=MODULE_ID, kind="section", label="Sanctions screening", parent_id="risk_compliance.workspace", order=30, ), ViewSurface( id="risk_compliance.sanctions.review", module_id=MODULE_ID, kind="section", label="Sanctions review queue", parent_id="risk_compliance.workspace", order=40, ), ViewSurface( id="risk_compliance.assurance.graph", module_id=MODULE_ID, kind="section", label="Assurance graph", parent_id="risk_compliance.workspace", order=50, ), ViewSurface( id="risk_compliance.action.import-snapshot", module_id=MODULE_ID, kind="action", label="Import sanctions source snapshot", parent_id="risk_compliance.sanctions.sources", order=60, ), ViewSurface( id="risk_compliance.action.run-screening", module_id=MODULE_ID, kind="action", label="Run sanctions screening", parent_id="risk_compliance.sanctions.screening", order=70, ), ViewSurface( id="risk_compliance.review.disposition", module_id=MODULE_ID, kind="dialog", label="Record screening disposition", parent_id="risk_compliance.sanctions.review", order=80, ), ViewSurface( id="risk_compliance.assurance.editor", module_id=MODULE_ID, kind="dialog", label="Assurance object editor", parent_id="risk_compliance.assurance.graph", order=90, ), ViewSurface( id="risk_compliance.action.connect-assurance", module_id=MODULE_ID, kind="action", label="Connect assurance objects", parent_id="risk_compliance.assurance.graph", order=100, ), ), ), tenant_summary_providers=(_tenant_summary,), architecture=ARCHITECTURE, migration_spec=MigrationSpec( module_id=MODULE_ID, metadata=Base.metadata, script_location=str(Path(__file__).with_name("migrations") / "versions"), migration_after=("connectors",), retirement_supported=True, retirement_provider=drop_table_retirement_provider( *_PERSISTENT_MODELS, label="Risk Compliance", ), retirement_notes=( "Destructive retirement removes immutable assurance graph, " "sanctions list, screening, and review evidence after a database " "snapshot." ), ), uninstall_guard_providers=( persistent_table_uninstall_guard( *_PERSISTENT_MODELS, label="Risk Compliance", ), ), documentation=DOCUMENTATION, ) def get_manifest() -> ModuleManifest: return manifest __all__ = [ "ADMIN_SCOPE", "MODULE_ID", "MODULE_VERSION", "PERMISSIONS", "READ_SCOPE", "ROLE_TEMPLATES", "SANCTIONS_ADMIN_SCOPE", "SANCTIONS_READ_SCOPE", "SANCTIONS_REVIEW_SCOPE", "SANCTIONS_SCREEN_SCOPE", "WRITE_SCOPE", "get_manifest", "manifest", ]