From a23b53dc9eed6c375a10f98c18f2e4e49215fb45 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sat, 1 Aug 2026 17:48:39 +0200 Subject: [PATCH] feat: implement assurance graph and screening evidence --- AGENTS.md | 6 + README.md | 11 +- docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md | 51 + .../backend/assurance.py | 1140 +++++++++++++++++ .../backend/db/models.py | 267 ++++ .../backend/manifest.py | 158 ++- .../b9c0d1e2f3a4_v0114_assurance_graph.py | 180 +++ .../backend/router.py | 365 ++++++ .../backend/schemas.py | 117 ++ .../backend/screening.py | 45 +- .../backend/search_source.py | 181 +++ tests/test_assurance_graph.py | 378 ++++++ tests/test_manifest.py | 5 + tests/test_migrations.py | 4 +- tests/test_sanctions_screening.py | 29 +- webui/src/api/riskCompliance.ts | 182 +++ .../riskCompliance/RiskCompliancePage.tsx | 721 ++++++++++- webui/src/styles/risk-compliance.css | 207 ++- 18 files changed, 4005 insertions(+), 42 deletions(-) create mode 100644 src/govoplan_risk_compliance/backend/assurance.py create mode 100644 src/govoplan_risk_compliance/backend/migrations/versions/b9c0d1e2f3a4_v0114_assurance_graph.py create mode 100644 src/govoplan_risk_compliance/backend/search_source.py create mode 100644 tests/test_assurance_graph.py diff --git a/AGENTS.md b/AGENTS.md index d31def9..66b8e25 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,5 +1,11 @@ # GovOPlaN Risk Compliance Codex Guide +## Documentation Contract + +- Treat documentation as part of every behavior change. Update this module's manifest-driven `DocumentationTopic` contributions for affected user and administrator behavior. +- Keep feature content here; `govoplan-docs` projects it without importing Risk Compliance internals. +- Maintain a static user/admin baseline and run `/mnt/DATA/git/govoplan/tools/checks/check-manifest-shapes.py` after behavior or manifest changes. + ## Scope This repository owns the GovOPlaN Risk Compliance platform module seed. diff --git a/README.md b/README.md index c3cd5aa..034aa08 100644 --- a/README.md +++ b/README.md @@ -12,9 +12,12 @@ The first complete vertical is sanctions screening. Connectors acquires immutabl The module includes database migrations, tenant-isolated APIs, an operational WebUI, append-only dispositions, time-bounded false-positive exceptions, -freshness reconciliation, explicit consumer gates, and audit events. Queue and -audit summaries deliberately retain only the minimum subject data needed for -the workflow. +freshness reconciliation, explicit consumer gates, and audit events. It also +provides a revisioned horizontal assurance graph from obligation through +effectiveness review, with bounded traversal, opaque governed-object links, +search, and optimistic concurrency. Queue, search, audit, and aggregate +summaries deliberately retain only the minimum subject data needed for their +purpose. ## Initial Ownership @@ -27,6 +30,8 @@ the workflow. - immutable sanctions list catalogues - sanctions screening and reviewer dispositions - stable screening evidence references and freshness gates +- effective-dated obligations, risks, controls, evidence, findings, corrective + measures, and effectiveness reviews ## Boundaries diff --git a/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md b/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md index db67094..7dbcbf5 100644 --- a/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md +++ b/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md @@ -16,12 +16,17 @@ Risk and compliance workflows for data protection incidents, DPIAs, compliance c - version-pinned screening runs and candidate explanations - reviewer dispositions, exceptions, and screening freshness - block, review, and degraded screening-gate decisions +- reusable assurance relationships from obligation through governed object, + risk, control, evidence, finding, corrective measure, and effectiveness + review ## Does Not Own - immutable audit log storage - records retention engine - inspection fieldwork +- domain-object lifecycle and corrective execution owned by the affected module +- policy rule evaluation or immutable audit-log storage ## Integration Candidates @@ -66,3 +71,49 @@ record. Fuzzy candidates always require review. A confirmed match blocks; a set of independently cleared false positives allows; source or execution uncertainty uses the consuming module's configured failure policy. + +## Horizontal Assurance Graph + +Sanctions screening is the first complete assurance vertical, not the whole +module model. Risk Compliance persists the reusable graph: + +```text +Obligation -> governed object -> risk -> control -> evidence -> finding -> measure -> effectiveness review +``` + +Every node and relationship has a stable tenant-local identity, immutable +revision history, effective and recorded time, owner and optional scope, +provenance, legal/policy/evidence references, and optimistic-concurrency +guards. Corrections create a new revision; the historical assertion is not +rewritten or deleted. Governed-object nodes retain only an opaque reference to +the object owned by another module. + +The tenant API exposes bounded current-object listing, revision histories, +aggregate counts, and graph traversal. Traversal is intentionally limited to +eight relationships and 500 returned objects. Search indexes only current +revisions, rechecks tenant access at result time, and omits confidential +descriptions. Aggregate responses contain counts, never labels or opaque +references. + +Completed sanctions screenings automatically project obligation, party, +exposure risk, version-pinned control, immutable evidence, and finding nodes. +This projection is idempotent and includes unsuccessful or stale screening +outcomes so operational uncertainty remains visible. The non-sanctions test +fixture proves the entire chain through corrective measure and effectiveness +review without coupling the graph to sanctions data. + +Policy may advise or block based on assurance state; Audit records events; +Files/Records retain referenced evidence; Tasks/Workflow may coordinate review +and correction. Risk Compliance owns risk, control, finding, and effectiveness +semantics, but the affected domain module remains authoritative for its object +and corrective execution. + +### Access, recovery, and retirement + +Read, write, and administration use the Risk Compliance workspace scopes and +are always tenant-bound. Generated sanctions evidence additionally keeps the +sanctions permissions on its source workflow. Database backup and restore is +the recovery mechanism for the immutable graph; migration verification covers +both graph tables. Module retirement is destructive and therefore requires a +database snapshot before graph, source, screening, and review evidence is +removed. diff --git a/src/govoplan_risk_compliance/backend/assurance.py b/src/govoplan_risk_compliance/backend/assurance.py new file mode 100644 index 0000000..022c66e --- /dev/null +++ b/src/govoplan_risk_compliance/backend/assurance.py @@ -0,0 +1,1140 @@ +from __future__ import annotations + +from collections import Counter, deque +from dataclasses import dataclass, field +from datetime import UTC, datetime +import hashlib +import json +import re +from typing import Any, Mapping + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.db.base import utcnow +from govoplan_risk_compliance.backend.db.models import ( + RiskAssuranceEdge, + RiskAssuranceNode, + RiskScreeningRun, +) +from govoplan_risk_compliance.backend.permissions import ( + ADMIN_SCOPE, + READ_SCOPE, + WRITE_SCOPE, +) + + +ASSURANCE_NODE_KINDS = ( + "obligation", + "governed_object", + "risk", + "control", + "evidence", + "finding", + "corrective_measure", + "effectiveness_review", +) +ASSURANCE_EDGE_RELATIONS: Mapping[str, tuple[str, str]] = { + "applies_to": ("obligation", "governed_object"), + "exposes_risk": ("governed_object", "risk"), + "mitigated_by": ("risk", "control"), + "evidenced_by": ("control", "evidence"), + "results_in": ("evidence", "finding"), + "addressed_by": ("finding", "corrective_measure"), + "reviewed_by": ("corrective_measure", "effectiveness_review"), +} +ASSURANCE_EDGE_STATES = frozenset({"active", "suspended", "retired"}) +ASSURANCE_NODE_STATES: Mapping[str, frozenset[str]] = { + "obligation": frozenset({"active", "suspended", "retired"}), + "governed_object": frozenset({"active", "inactive", "retired"}), + "risk": frozenset({"identified", "assessed", "accepted", "mitigated", "closed"}), + "control": frozenset({"designed", "implemented", "effective", "failed", "suspended", "retired"}), + "evidence": frozenset({"current", "stale", "invalid", "superseded"}), + "finding": frozenset({"open", "accepted", "exception", "remediating", "resolved"}), + "corrective_measure": frozenset({"planned", "in_progress", "completed", "cancelled"}), + "effectiveness_review": frozenset({"pending", "effective", "ineffective", "inconclusive"}), +} + +MAX_GRAPH_DEPTH = 8 +MAX_GRAPH_ITEMS = 500 +MAX_PROVENANCE_BYTES = 100_000 +_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/-]{0,254}$") + + +class RiskAssuranceError(ValueError): + pass + + +class RiskAssuranceAccessError(RiskAssuranceError): + pass + + +class RiskAssuranceNotFoundError(RiskAssuranceError): + pass + + +class RiskAssuranceConflictError(RiskAssuranceError): + pass + + +@dataclass(frozen=True, slots=True) +class AssuranceNodeInput: + stable_id: str + kind: str + label: str + state: str + owner_ref: str + valid_from: datetime + description: str | None = None + scope_ref: str | None = None + governed_object_ref: str | None = None + valid_to: datetime | None = None + provenance: Mapping[str, Any] = field(default_factory=dict) + legal_basis_refs: tuple[str, ...] = () + policy_refs: tuple[str, ...] = () + evidence_refs: tuple[str, ...] = () + classification: str = "internal" + + def __post_init__(self) -> None: + _validate_node_input(self) + + +@dataclass(frozen=True, slots=True) +class AssuranceEdgeInput: + stable_id: str + source_node_ref: str + target_node_ref: str + relation: str + owner_ref: str + valid_from: datetime + state: str = "active" + scope_ref: str | None = None + valid_to: datetime | None = None + provenance: Mapping[str, Any] = field(default_factory=dict) + legal_basis_refs: tuple[str, ...] = () + policy_refs: tuple[str, ...] = () + evidence_refs: tuple[str, ...] = () + + def __post_init__(self) -> None: + _validate_edge_input(self) + + +@dataclass(frozen=True, slots=True) +class AssuranceGraph: + root_ref: str + nodes: tuple[RiskAssuranceNode, ...] + edges: tuple[RiskAssuranceEdge, ...] + truncated: bool + + +def create_assurance_node( + session: Session, + principal: ApiPrincipal, + *, + value: AssuranceNodeInput, +) -> RiskAssuranceNode: + _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + if _current_node(session, principal.tenant_id, value.stable_id) is not None: + raise RiskAssuranceConflictError( + f"Assurance node {value.stable_id!r} already exists." + ) + return _insert_node( + session, + tenant_id=principal.tenant_id, + value=value, + revision=1, + previous=None, + actor_id=_actor_id(principal), + ) + + +def revise_assurance_node( + session: Session, + principal: ApiPrincipal, + *, + stable_id: str, + expected_revision: int, + value: AssuranceNodeInput, +) -> RiskAssuranceNode: + _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + if value.stable_id != stable_id: + raise RiskAssuranceConflictError( + "The assurance node identity cannot change during revision." + ) + current = _locked_current_node(session, principal.tenant_id, stable_id) + if current is None: + raise RiskAssuranceNotFoundError("Assurance node not found.") + if current.revision != expected_revision: + raise RiskAssuranceConflictError( + f"Expected revision {expected_revision}, current revision is {current.revision}." + ) + if current.kind != value.kind: + raise RiskAssuranceConflictError( + "The assurance node kind cannot change during revision." + ) + return _insert_node( + session, + tenant_id=principal.tenant_id, + value=value, + revision=current.revision + 1, + previous=current, + actor_id=_actor_id(principal), + ) + + +def create_assurance_edge( + session: Session, + principal: ApiPrincipal, + *, + value: AssuranceEdgeInput, +) -> RiskAssuranceEdge: + _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + if _current_edge(session, principal.tenant_id, value.stable_id) is not None: + raise RiskAssuranceConflictError( + f"Assurance edge {value.stable_id!r} already exists." + ) + _validate_edge_endpoints(session, principal.tenant_id, value) + return _insert_edge( + session, + tenant_id=principal.tenant_id, + value=value, + revision=1, + previous=None, + actor_id=_actor_id(principal), + ) + + +def revise_assurance_edge( + session: Session, + principal: ApiPrincipal, + *, + stable_id: str, + expected_revision: int, + value: AssuranceEdgeInput, +) -> RiskAssuranceEdge: + _require_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + if value.stable_id != stable_id: + raise RiskAssuranceConflictError( + "The assurance edge identity cannot change during revision." + ) + current = _locked_current_edge(session, principal.tenant_id, stable_id) + if current is None: + raise RiskAssuranceNotFoundError("Assurance edge not found.") + if current.revision != expected_revision: + raise RiskAssuranceConflictError( + f"Expected revision {expected_revision}, current revision is {current.revision}." + ) + identity = ( + value.source_node_ref, + value.target_node_ref, + value.relation, + ) + if identity != ( + current.source_node_ref, + current.target_node_ref, + current.relation, + ): + raise RiskAssuranceConflictError( + "Edge endpoints and relation cannot change during revision." + ) + _validate_edge_endpoints(session, principal.tenant_id, value) + return _insert_edge( + session, + tenant_id=principal.tenant_id, + value=value, + revision=current.revision + 1, + previous=current, + actor_id=_actor_id(principal), + ) + + +def get_assurance_node( + session: Session, + principal: ApiPrincipal, + *, + stable_id: str, +) -> RiskAssuranceNode: + _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + item = _current_node(session, principal.tenant_id, stable_id) + if item is None: + raise RiskAssuranceNotFoundError("Assurance node not found.") + return item + + +def list_assurance_node_history( + session: Session, + principal: ApiPrincipal, + *, + stable_id: str, + limit: int = 100, +) -> tuple[RiskAssuranceNode, ...]: + _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + bounded_limit = _bounded_limit(limit) + return tuple( + session.scalars( + select(RiskAssuranceNode) + .where( + RiskAssuranceNode.tenant_id == principal.tenant_id, + RiskAssuranceNode.stable_id == stable_id, + ) + .order_by(RiskAssuranceNode.revision.desc()) + .limit(bounded_limit) + ) + ) + + +def list_assurance_edge_history( + session: Session, + principal: ApiPrincipal, + *, + stable_id: str, + limit: int = 100, +) -> tuple[RiskAssuranceEdge, ...]: + _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + bounded_limit = _bounded_limit(limit) + return tuple( + session.scalars( + select(RiskAssuranceEdge) + .where( + RiskAssuranceEdge.tenant_id == principal.tenant_id, + RiskAssuranceEdge.stable_id == stable_id, + ) + .order_by(RiskAssuranceEdge.revision.desc()) + .limit(bounded_limit) + ) + ) + + +def list_assurance_nodes( + session: Session, + principal: ApiPrincipal, + *, + kind: str | None = None, + state: str | None = None, + governed_object_ref: str | None = None, + query: str | None = None, + limit: int = 100, +) -> tuple[RiskAssuranceNode, ...]: + _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + statement = select(RiskAssuranceNode).where( + RiskAssuranceNode.tenant_id == principal.tenant_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + if kind: + _validate_kind(kind) + statement = statement.where(RiskAssuranceNode.kind == kind) + if state: + statement = statement.where(RiskAssuranceNode.state == state) + if governed_object_ref: + statement = statement.where( + RiskAssuranceNode.governed_object_ref == governed_object_ref + ) + search = str(query or "").strip() + if search: + pattern = f"%{_escape_like(search)}%" + statement = statement.where( + or_( + RiskAssuranceNode.label.ilike(pattern, escape="\\"), + RiskAssuranceNode.description.ilike(pattern, escape="\\"), + RiskAssuranceNode.stable_id.ilike(pattern, escape="\\"), + ) + ) + return tuple( + session.scalars( + statement.order_by( + RiskAssuranceNode.kind, + RiskAssuranceNode.label, + RiskAssuranceNode.stable_id, + ).limit(_bounded_limit(limit)) + ) + ) + + +def assurance_graph( + session: Session, + principal: ApiPrincipal, + *, + root_ref: str, + max_depth: int = 4, + limit: int = 200, +) -> AssuranceGraph: + _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + if not 0 <= max_depth <= MAX_GRAPH_DEPTH: + raise RiskAssuranceConflictError( + f"Graph depth must be between 0 and {MAX_GRAPH_DEPTH}." + ) + bounded_limit = _bounded_limit(limit) + root = _current_node(session, principal.tenant_id, root_ref) + if root is None: + raise RiskAssuranceNotFoundError("Assurance graph root not found.") + + nodes: dict[str, RiskAssuranceNode] = {root.stable_id: root} + edges: dict[str, RiskAssuranceEdge] = {} + queue: deque[tuple[str, int]] = deque(((root.stable_id, 0),)) + expanded: set[str] = set() + truncated = False + while queue: + node_ref, depth = queue.popleft() + if node_ref in expanded or depth >= max_depth: + continue + expanded.add(node_ref) + remaining = bounded_limit - len(edges) + if remaining <= 0: + truncated = True + break + connected = tuple( + session.scalars( + select(RiskAssuranceEdge) + .where( + RiskAssuranceEdge.tenant_id == principal.tenant_id, + RiskAssuranceEdge.superseded_at.is_(None), + RiskAssuranceEdge.state == "active", + or_( + RiskAssuranceEdge.source_node_ref == node_ref, + RiskAssuranceEdge.target_node_ref == node_ref, + ), + ) + .order_by(RiskAssuranceEdge.stable_id) + .limit(remaining + 1) + ) + ) + if len(connected) > remaining: + connected = connected[:remaining] + truncated = True + neighbor_refs: set[str] = set() + for edge in connected: + edges[edge.stable_id] = edge + neighbor_refs.update( + (edge.source_node_ref, edge.target_node_ref) + ) + missing = neighbor_refs - nodes.keys() + if missing: + for item in session.scalars( + select(RiskAssuranceNode).where( + RiskAssuranceNode.tenant_id == principal.tenant_id, + RiskAssuranceNode.superseded_at.is_(None), + RiskAssuranceNode.stable_id.in_(missing), + ) + ): + nodes[item.stable_id] = item + for neighbor in sorted(neighbor_refs): + if neighbor not in expanded: + queue.append((neighbor, depth + 1)) + if len(nodes) >= bounded_limit: + truncated = truncated or bool(queue) + break + selected_nodes = tuple( + sorted(nodes.values(), key=lambda item: (item.kind, item.stable_id)) + )[:bounded_limit] + selected_refs = {item.stable_id for item in selected_nodes} + selected_edges = tuple( + edge + for edge in sorted(edges.values(), key=lambda item: item.stable_id) + if edge.source_node_ref in selected_refs + and edge.target_node_ref in selected_refs + )[:bounded_limit] + return AssuranceGraph( + root_ref=root_ref, + nodes=selected_nodes, + edges=selected_edges, + truncated=truncated, + ) + + +def assurance_summary( + session: Session, + principal: ApiPrincipal, +) -> dict[str, object]: + _require_scope(principal, READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + rows = session.execute( + select( + RiskAssuranceNode.kind, + RiskAssuranceNode.state, + func.count(RiskAssuranceNode.id), + ) + .where( + RiskAssuranceNode.tenant_id == principal.tenant_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + .group_by(RiskAssuranceNode.kind, RiskAssuranceNode.state) + ) + by_kind: Counter[str] = Counter() + by_state: Counter[str] = Counter() + for kind, state, count in rows: + by_kind[str(kind)] += int(count) + by_state[str(state)] += int(count) + return { + "node_count": sum(by_kind.values()), + "edge_count": int( + session.scalar( + select(func.count(RiskAssuranceEdge.id)).where( + RiskAssuranceEdge.tenant_id == principal.tenant_id, + RiskAssuranceEdge.superseded_at.is_(None), + ) + ) + or 0 + ), + "by_kind": dict(sorted(by_kind.items())), + "by_state": dict(sorted(by_state.items())), + } + + +def record_sanctions_assurance( + session: Session, + principal: ApiPrincipal, + *, + run: RiskScreeningRun, +) -> tuple[RiskAssuranceNode, ...]: + """Project a completed sanctions run into the horizontal assurance graph.""" + if run.tenant_id != principal.tenant_id: + raise RiskAssuranceAccessError("Screening assurance cannot cross tenants.") + if run.status != "completed": + raise RiskAssuranceConflictError( + "Only completed screenings can become assurance evidence." + ) + source = run.list_snapshot + subject = run.subject_snapshot + now = _as_aware(run.completed_at or utcnow()) + started_at = _as_aware(run.started_at) + source_effective_at = _as_aware( + source.effective_at or source.acquired_at + ) + actor_id = _actor_id(principal) + legal_ref = ( + f"sanctions-list:{source.jurisdiction}:{source.source_id}:" + f"{source.source_version}" + ) + evidence_ref = f"risk-compliance:sanctions-screening:{run.id}" + owner_ref = "risk-compliance:sanctions" + inputs = ( + AssuranceNodeInput( + stable_id=_bounded_id( + f"sanctions-obligation:{source.jurisdiction}:{source.list_type}" + ), + kind="obligation", + label=f"{source.jurisdiction} sanctions screening obligation", + state="active", + owner_ref=owner_ref, + valid_from=source_effective_at, + provenance={ + "provider_id": source.provider_id, + "source_version": source.source_version, + }, + legal_basis_refs=(legal_ref,), + ), + AssuranceNodeInput( + stable_id=f"sanctions-subject:{subject.fingerprint}", + kind="governed_object", + label="Screened party", + state="active", + owner_ref=owner_ref, + valid_from=started_at, + governed_object_ref=( + subject.subject_ref + or f"risk-compliance:screening-subject:{subject.id}" + ), + provenance={"subject_fingerprint": subject.fingerprint}, + classification="confidential", + ), + AssuranceNodeInput( + stable_id=f"sanctions-risk:{subject.fingerprint}", + kind="risk", + label="Sanctions exposure", + state=("identified" if run.outcome == "potential" else "mitigated"), + owner_ref=owner_ref, + valid_from=started_at, + provenance={"screening_outcome": run.outcome}, + legal_basis_refs=(legal_ref,), + classification="confidential", + ), + AssuranceNodeInput( + stable_id=_bounded_id(f"sanctions-control:{source.provider_id}"), + kind="control", + label="Version-pinned sanctions screening", + state=( + "failed" + if run.outcome in {"unavailable", "stale", "insufficient"} + else "effective" + ), + owner_ref=owner_ref, + valid_from=started_at, + provenance={ + "matcher_version": run.matcher_version, + "normalization_version": run.normalization_version, + "policy_version": run.policy_version, + }, + policy_refs=(f"sanctions-policy:{run.policy_version}",), + ), + AssuranceNodeInput( + stable_id=f"sanctions-evidence:{run.id}", + kind="evidence", + label="Sanctions screening evidence", + state=( + "stale" + if run.outcome == "stale" + else "invalid" + if run.outcome in {"unavailable", "insufficient"} + else "current" + ), + owner_ref=owner_ref, + valid_from=now, + provenance={ + "list_snapshot_id": run.list_snapshot_id, + "list_source_version": source.source_version, + "request_hash": run.request_hash, + }, + evidence_refs=(evidence_ref,), + classification="confidential", + ), + AssuranceNodeInput( + stable_id=f"sanctions-finding:{run.id}", + kind="finding", + label=( + "Potential sanctions match" + if run.outcome == "potential" + else "Sanctions screening result" + ), + state="open" if run.outcome == "potential" else "resolved", + owner_ref=owner_ref, + valid_from=now, + provenance={ + "outcome": run.outcome, + "candidate_count": run.candidate_count, + }, + legal_basis_refs=(legal_ref,), + evidence_refs=(evidence_ref,), + classification="confidential", + ), + ) + nodes = tuple( + _ensure_node( + session, + tenant_id=principal.tenant_id, + value=value, + actor_id=actor_id, + ) + for value in inputs + ) + edge_inputs = ( + (nodes[0], "applies_to", nodes[1]), + (nodes[1], "exposes_risk", nodes[2]), + (nodes[2], "mitigated_by", nodes[3]), + (nodes[3], "evidenced_by", nodes[4]), + (nodes[4], "results_in", nodes[5]), + ) + for source_node, relation, target_node in edge_inputs: + _ensure_edge( + session, + tenant_id=principal.tenant_id, + value=AssuranceEdgeInput( + stable_id=_edge_id( + source_node.stable_id, + relation, + target_node.stable_id, + ), + source_node_ref=source_node.stable_id, + target_node_ref=target_node.stable_id, + relation=relation, + owner_ref=owner_ref, + valid_from=now, + provenance={"screening_run_id": run.id}, + evidence_refs=(evidence_ref,), + ), + actor_id=actor_id, + ) + session.flush() + return nodes + + +def _insert_node( + session: Session, + *, + tenant_id: str, + value: AssuranceNodeInput, + revision: int, + previous: RiskAssuranceNode | None, + actor_id: str | None, +) -> RiskAssuranceNode: + recorded_at = utcnow() + if previous is not None: + previous.superseded_at = recorded_at + item = RiskAssuranceNode( + tenant_id=tenant_id, + stable_id=value.stable_id, + kind=value.kind, + revision=revision, + previous_revision_id=previous.id if previous is not None else None, + label=value.label.strip(), + description=_optional_text(value.description), + state=value.state, + owner_ref=value.owner_ref.strip(), + scope_ref=_optional_text(value.scope_ref), + governed_object_ref=_optional_text(value.governed_object_ref), + valid_from=value.valid_from, + valid_to=value.valid_to, + recorded_at=recorded_at, + provenance=dict(value.provenance), + legal_basis_refs=list(value.legal_basis_refs), + policy_refs=list(value.policy_refs), + evidence_refs=list(value.evidence_refs), + classification=value.classification.strip(), + created_by=actor_id, + ) + session.add(item) + session.flush() + return item + + +def _insert_edge( + session: Session, + *, + tenant_id: str, + value: AssuranceEdgeInput, + revision: int, + previous: RiskAssuranceEdge | None, + actor_id: str | None, +) -> RiskAssuranceEdge: + recorded_at = utcnow() + if previous is not None: + previous.superseded_at = recorded_at + item = RiskAssuranceEdge( + tenant_id=tenant_id, + stable_id=value.stable_id, + revision=revision, + previous_revision_id=previous.id if previous is not None else None, + source_node_ref=value.source_node_ref, + target_node_ref=value.target_node_ref, + relation=value.relation, + state=value.state, + owner_ref=value.owner_ref.strip(), + scope_ref=_optional_text(value.scope_ref), + valid_from=value.valid_from, + valid_to=value.valid_to, + recorded_at=recorded_at, + provenance=dict(value.provenance), + legal_basis_refs=list(value.legal_basis_refs), + policy_refs=list(value.policy_refs), + evidence_refs=list(value.evidence_refs), + created_by=actor_id, + ) + session.add(item) + session.flush() + return item + + +def _ensure_node( + session: Session, + *, + tenant_id: str, + value: AssuranceNodeInput, + actor_id: str | None, +) -> RiskAssuranceNode: + current = _locked_current_node(session, tenant_id, value.stable_id) + if current is None: + return _insert_node( + session, + tenant_id=tenant_id, + value=value, + revision=1, + previous=None, + actor_id=actor_id, + ) + if _node_matches(current, value): + return current + if current.kind != value.kind: + raise RiskAssuranceConflictError( + f"Assurance node {value.stable_id!r} changed kind." + ) + return _insert_node( + session, + tenant_id=tenant_id, + value=value, + revision=current.revision + 1, + previous=current, + actor_id=actor_id, + ) + + +def _ensure_edge( + session: Session, + *, + tenant_id: str, + value: AssuranceEdgeInput, + actor_id: str | None, +) -> RiskAssuranceEdge: + current = _locked_current_edge(session, tenant_id, value.stable_id) + if current is None: + _validate_edge_endpoints(session, tenant_id, value) + return _insert_edge( + session, + tenant_id=tenant_id, + value=value, + revision=1, + previous=None, + actor_id=actor_id, + ) + if _edge_matches(current, value): + return current + identity = ( + value.source_node_ref, + value.target_node_ref, + value.relation, + ) + if identity != ( + current.source_node_ref, + current.target_node_ref, + current.relation, + ): + raise RiskAssuranceConflictError( + f"Assurance edge {value.stable_id!r} changed identity." + ) + return _insert_edge( + session, + tenant_id=tenant_id, + value=value, + revision=current.revision + 1, + previous=current, + actor_id=actor_id, + ) + + +def _current_node( + session: Session, + tenant_id: str, + stable_id: str, +) -> RiskAssuranceNode | None: + return session.scalar( + select(RiskAssuranceNode).where( + RiskAssuranceNode.tenant_id == tenant_id, + RiskAssuranceNode.stable_id == stable_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + ) + + +def _locked_current_node( + session: Session, + tenant_id: str, + stable_id: str, +) -> RiskAssuranceNode | None: + return session.scalar( + select(RiskAssuranceNode) + .where( + RiskAssuranceNode.tenant_id == tenant_id, + RiskAssuranceNode.stable_id == stable_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + .with_for_update() + ) + + +def _current_edge( + session: Session, + tenant_id: str, + stable_id: str, +) -> RiskAssuranceEdge | None: + return session.scalar( + select(RiskAssuranceEdge).where( + RiskAssuranceEdge.tenant_id == tenant_id, + RiskAssuranceEdge.stable_id == stable_id, + RiskAssuranceEdge.superseded_at.is_(None), + ) + ) + + +def _locked_current_edge( + session: Session, + tenant_id: str, + stable_id: str, +) -> RiskAssuranceEdge | None: + return session.scalar( + select(RiskAssuranceEdge) + .where( + RiskAssuranceEdge.tenant_id == tenant_id, + RiskAssuranceEdge.stable_id == stable_id, + RiskAssuranceEdge.superseded_at.is_(None), + ) + .with_for_update() + ) + + +def _validate_edge_endpoints( + session: Session, + tenant_id: str, + value: AssuranceEdgeInput, +) -> None: + source = _current_node(session, tenant_id, value.source_node_ref) + target = _current_node(session, tenant_id, value.target_node_ref) + if source is None or target is None: + raise RiskAssuranceNotFoundError( + "Both assurance edge endpoints must exist in the current tenant." + ) + expected = ASSURANCE_EDGE_RELATIONS[value.relation] + if (source.kind, target.kind) != expected: + raise RiskAssuranceConflictError( + f"Relation {value.relation!r} requires {expected[0]} -> {expected[1]}." + ) + + +def _validate_node_input(value: AssuranceNodeInput) -> None: + _validate_id(value.stable_id, "Assurance node stable id") + _validate_kind(value.kind) + _require_text(value.label, "Assurance node label", 500) + _require_text(value.owner_ref, "Assurance node owner", 500) + _validate_optional_text(value.description, "Assurance node description", 20_000) + _validate_optional_text(value.scope_ref, "Assurance node scope", 500) + _validate_optional_text( + value.governed_object_ref, + "Assurance governed object reference", + 1_000, + ) + if value.state not in ASSURANCE_NODE_STATES[value.kind]: + raise RiskAssuranceError( + f"Unsupported {value.kind} state: {value.state!r}." + ) + if value.kind == "governed_object" and not _optional_text( + value.governed_object_ref + ): + raise RiskAssuranceError( + "Governed-object nodes require an opaque governed object reference." + ) + _validate_period(value.valid_from, value.valid_to) + _validate_reference_values(value.legal_basis_refs, "legal basis") + _validate_reference_values(value.policy_refs, "policy") + _validate_reference_values(value.evidence_refs, "evidence") + _validate_provenance(value.provenance) + _require_text(value.classification, "Assurance classification", 50) + + +def _validate_edge_input(value: AssuranceEdgeInput) -> None: + _validate_id(value.stable_id, "Assurance edge stable id") + _validate_id(value.source_node_ref, "Assurance edge source") + _validate_id(value.target_node_ref, "Assurance edge target") + if value.source_node_ref == value.target_node_ref: + raise RiskAssuranceError("Assurance edges cannot reference themselves.") + if value.relation not in ASSURANCE_EDGE_RELATIONS: + raise RiskAssuranceError( + f"Unsupported assurance relation: {value.relation!r}." + ) + if value.state not in ASSURANCE_EDGE_STATES: + raise RiskAssuranceError( + f"Unsupported assurance edge state: {value.state!r}." + ) + _require_text(value.owner_ref, "Assurance edge owner", 500) + _validate_optional_text(value.scope_ref, "Assurance edge scope", 500) + _validate_period(value.valid_from, value.valid_to) + _validate_reference_values(value.legal_basis_refs, "legal basis") + _validate_reference_values(value.policy_refs, "policy") + _validate_reference_values(value.evidence_refs, "evidence") + _validate_provenance(value.provenance) + + +def _validate_kind(value: str) -> None: + if value not in ASSURANCE_NODE_STATES: + raise RiskAssuranceError(f"Unsupported assurance node kind: {value!r}.") + + +def _validate_period(valid_from: datetime, valid_to: datetime | None) -> None: + for value in (valid_from, valid_to): + if value is not None and ( + value.tzinfo is None or value.utcoffset() is None + ): + raise RiskAssuranceError( + "Assurance effective times must include a timezone." + ) + if valid_to is not None and valid_to <= valid_from: + raise RiskAssuranceError( + "Assurance valid_to must be later than valid_from." + ) + + +def _validate_reference_values(values: tuple[str, ...], label: str) -> None: + if len(values) > 100: + raise RiskAssuranceError( + f"Assurance {label} references are limited to 100 values." + ) + if len(values) != len(set(values)): + raise RiskAssuranceError( + f"Assurance {label} references cannot contain duplicates." + ) + for value in values: + _require_text(value, f"Assurance {label} reference", 1000) + + +def _node_matches(item: RiskAssuranceNode, value: AssuranceNodeInput) -> bool: + return ( + item.kind, + item.label, + item.description, + item.state, + item.owner_ref, + item.scope_ref, + item.governed_object_ref, + _time_key(item.valid_from), + _time_key(item.valid_to), + dict(item.provenance), + tuple(item.legal_basis_refs), + tuple(item.policy_refs), + tuple(item.evidence_refs), + item.classification, + ) == ( + value.kind, + value.label.strip(), + _optional_text(value.description), + value.state, + value.owner_ref.strip(), + _optional_text(value.scope_ref), + _optional_text(value.governed_object_ref), + _time_key(value.valid_from), + _time_key(value.valid_to), + dict(value.provenance), + value.legal_basis_refs, + value.policy_refs, + value.evidence_refs, + value.classification.strip(), + ) + + +def _edge_matches(item: RiskAssuranceEdge, value: AssuranceEdgeInput) -> bool: + return ( + item.source_node_ref, + item.target_node_ref, + item.relation, + item.state, + item.owner_ref, + item.scope_ref, + _time_key(item.valid_from), + _time_key(item.valid_to), + dict(item.provenance), + tuple(item.legal_basis_refs), + tuple(item.policy_refs), + tuple(item.evidence_refs), + ) == ( + value.source_node_ref, + value.target_node_ref, + value.relation, + value.state, + value.owner_ref.strip(), + _optional_text(value.scope_ref), + _time_key(value.valid_from), + _time_key(value.valid_to), + dict(value.provenance), + value.legal_basis_refs, + value.policy_refs, + value.evidence_refs, + ) + + +def _edge_id(source_ref: str, relation: str, target_ref: str) -> str: + digest = hashlib.sha256( + f"{source_ref}\0{relation}\0{target_ref}".encode("utf-8") + ).hexdigest()[:32] + return f"assurance-edge:{digest}" + + +def _as_aware(value: datetime) -> datetime: + return value.replace(tzinfo=UTC) if value.tzinfo is None else value + + +def _time_key(value: datetime | None) -> datetime | None: + return _as_aware(value).astimezone(UTC) if value is not None else None + + +def _bounded_id(value: str) -> str: + if len(value) <= 255 and _ID_PATTERN.fullmatch(value): + return value + digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:32] + prefix = re.sub(r"[^A-Za-z0-9_.:/-]", "-", value[:200]).rstrip("-") + return f"{prefix}:{digest}"[:255] + + +def _validate_id(value: str, label: str) -> None: + if not _ID_PATTERN.fullmatch(str(value or "")): + raise RiskAssuranceError(f"{label} is invalid.") + + +def _require_text(value: str, label: str, limit: int) -> None: + cleaned = str(value or "").strip() + if not cleaned: + raise RiskAssuranceError(f"{label} is required.") + if len(cleaned) > limit: + raise RiskAssuranceError(f"{label} is limited to {limit} characters.") + + +def _optional_text(value: str | None) -> str | None: + cleaned = str(value or "").strip() + return cleaned or None + + +def _validate_optional_text(value: str | None, label: str, limit: int) -> None: + cleaned = _optional_text(value) + if cleaned is not None and len(cleaned) > limit: + raise RiskAssuranceError(f"{label} is limited to {limit} characters.") + + +def _validate_provenance(value: Mapping[str, Any]) -> None: + try: + encoded = json.dumps( + dict(value), + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise RiskAssuranceError( + "Assurance provenance must contain JSON-compatible values." + ) from exc + if len(encoded) > MAX_PROVENANCE_BYTES: + raise RiskAssuranceError( + f"Assurance provenance is limited to {MAX_PROVENANCE_BYTES} bytes." + ) + + +def _bounded_limit(value: int) -> int: + if not 1 <= value <= MAX_GRAPH_ITEMS: + raise RiskAssuranceConflictError( + f"Assurance result limits must be between 1 and {MAX_GRAPH_ITEMS}." + ) + return value + + +def _escape_like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def _require_scope(principal: ApiPrincipal, *scopes: str) -> None: + if not any(principal.has(scope) for scope in scopes): + raise RiskAssuranceAccessError("Risk assurance access is not permitted.") + + +def _actor_id(principal: ApiPrincipal) -> str | None: + return getattr(principal.user, "id", None) or getattr( + principal.account, + "id", + None, + ) + + +__all__ = [ + "ASSURANCE_EDGE_RELATIONS", + "ASSURANCE_NODE_KINDS", + "AssuranceEdgeInput", + "AssuranceGraph", + "AssuranceNodeInput", + "RiskAssuranceAccessError", + "RiskAssuranceConflictError", + "RiskAssuranceError", + "RiskAssuranceNotFoundError", + "assurance_graph", + "assurance_summary", + "create_assurance_edge", + "create_assurance_node", + "get_assurance_node", + "list_assurance_edge_history", + "list_assurance_node_history", + "list_assurance_nodes", + "record_sanctions_assurance", + "revise_assurance_edge", + "revise_assurance_node", +] diff --git a/src/govoplan_risk_compliance/backend/db/models.py b/src/govoplan_risk_compliance/backend/db/models.py index c944665..0e467a8 100644 --- a/src/govoplan_risk_compliance/backend/db/models.py +++ b/src/govoplan_risk_compliance/backend/db/models.py @@ -841,7 +841,274 @@ class RiskScreeningException(Base, TimestampMixin): ) +class RiskAssuranceNode(Base, TimestampMixin): + __tablename__ = "risk_assurance_nodes" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "stable_id", + "revision", + name="uq_risk_assurance_node_revision", + ), + Index( + "ix_risk_assurance_node_current", + "tenant_id", + "kind", + "state", + "superseded_at", + ), + Index( + "ix_risk_assurance_node_governed_object", + "tenant_id", + "governed_object_ref", + "superseded_at", + ), + ) + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=new_uuid, + ) + tenant_id: Mapped[str] = mapped_column( + String(36), + nullable=False, + index=True, + ) + stable_id: Mapped[str] = mapped_column( + String(255), + nullable=False, + index=True, + ) + kind: Mapped[str] = mapped_column( + String(40), + nullable=False, + index=True, + ) + revision: Mapped[int] = mapped_column( + Integer, + nullable=False, + ) + previous_revision_id: Mapped[str | None] = mapped_column( + ForeignKey( + "risk_assurance_nodes.id", + ondelete="RESTRICT", + ), + nullable=True, + ) + label: Mapped[str] = mapped_column( + String(500), + nullable=False, + ) + description: Mapped[str | None] = mapped_column( + Text, + nullable=True, + ) + state: Mapped[str] = mapped_column( + String(40), + nullable=False, + index=True, + ) + owner_ref: Mapped[str] = mapped_column( + String(500), + nullable=False, + index=True, + ) + scope_ref: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + index=True, + ) + governed_object_ref: Mapped[str | None] = mapped_column( + String(1000), + nullable=True, + index=True, + ) + valid_from: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + index=True, + ) + valid_to: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + recorded_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + index=True, + ) + superseded_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + legal_basis_refs: Mapped[list[str]] = mapped_column( + JSON, + default=list, + nullable=False, + ) + policy_refs: Mapped[list[str]] = mapped_column( + JSON, + default=list, + nullable=False, + ) + evidence_refs: Mapped[list[str]] = mapped_column( + JSON, + default=list, + nullable=False, + ) + classification: Mapped[str] = mapped_column( + String(50), + default="internal", + nullable=False, + index=True, + ) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + + +class RiskAssuranceEdge(Base, TimestampMixin): + __tablename__ = "risk_assurance_edges" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "stable_id", + "revision", + name="uq_risk_assurance_edge_revision", + ), + Index( + "ix_risk_assurance_edge_source", + "tenant_id", + "source_node_ref", + "state", + "superseded_at", + ), + Index( + "ix_risk_assurance_edge_target", + "tenant_id", + "target_node_ref", + "state", + "superseded_at", + ), + ) + + id: Mapped[str] = mapped_column( + String(36), + primary_key=True, + default=new_uuid, + ) + tenant_id: Mapped[str] = mapped_column( + String(36), + nullable=False, + index=True, + ) + stable_id: Mapped[str] = mapped_column( + String(255), + nullable=False, + index=True, + ) + revision: Mapped[int] = mapped_column( + Integer, + nullable=False, + ) + previous_revision_id: Mapped[str | None] = mapped_column( + ForeignKey( + "risk_assurance_edges.id", + ondelete="RESTRICT", + ), + nullable=True, + ) + source_node_ref: Mapped[str] = mapped_column( + String(255), + nullable=False, + index=True, + ) + target_node_ref: Mapped[str] = mapped_column( + String(255), + nullable=False, + index=True, + ) + relation: Mapped[str] = mapped_column( + String(50), + nullable=False, + index=True, + ) + state: Mapped[str] = mapped_column( + String(30), + default="active", + nullable=False, + index=True, + ) + owner_ref: Mapped[str] = mapped_column( + String(500), + nullable=False, + index=True, + ) + scope_ref: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + index=True, + ) + valid_from: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + index=True, + ) + valid_to: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + recorded_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + index=True, + ) + superseded_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + legal_basis_refs: Mapped[list[str]] = mapped_column( + JSON, + default=list, + nullable=False, + ) + policy_refs: Mapped[list[str]] = mapped_column( + JSON, + default=list, + nullable=False, + ) + evidence_refs: Mapped[list[str]] = mapped_column( + JSON, + default=list, + nullable=False, + ) + created_by: Mapped[str | None] = mapped_column( + String(255), + nullable=True, + index=True, + ) + + __all__ = [ + "RiskAssuranceEdge", + "RiskAssuranceNode", "RiskSanctionsAddress", "RiskSanctionsAlias", "RiskSanctionsDate", diff --git a/src/govoplan_risk_compliance/backend/manifest.py b/src/govoplan_risk_compliance/backend/manifest.py index 82f439f..6b2681d 100644 --- a/src/govoplan_risk_compliance/backend/manifest.py +++ b/src/govoplan_risk_compliance/backend/manifest.py @@ -26,9 +26,17 @@ from govoplan_core.core.modules import ( 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, @@ -66,6 +74,8 @@ OPTIONAL_DEPENDENCIES = ( "connectors", ) _PERSISTENT_MODELS = ( + RiskAssuranceEdge, + RiskAssuranceNode, RiskScreeningException, RiskScreeningDisposition, RiskScreeningCandidate, @@ -79,6 +89,72 @@ _PERSISTENT_MODELS = ( 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, @@ -140,9 +216,7 @@ ROLE_TEMPLATES = ( RoleTemplate( slug="risk_compliance_manager", name="Risk Compliance manager", - description=( - "Manage compliance workflows and administer sanctions screening." - ), + description=("Manage compliance workflows and administer sanctions screening."), permissions=( READ_SCOPE, WRITE_SCOPE, @@ -155,9 +229,7 @@ ROLE_TEMPLATES = ( RoleTemplate( slug="risk_compliance_reviewer", name="Risk Compliance reviewer", - description=( - "Run screenings and independently review potential matches." - ), + description=("Run screenings and independently review potential matches."), permissions=( READ_SCOPE, SANCTIONS_READ_SCOPE, @@ -188,13 +260,19 @@ def _sanctions_screening_provider(_context): 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 - ) + .filter(RiskSanctionsListSnapshot.tenant_id == tenant_id) .count() ), "risk_screening_runs": ( @@ -212,6 +290,14 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]: ) .count() ), + "risk_assurance_nodes": ( + session.query(RiskAssuranceNode) + .filter( + RiskAssuranceNode.tenant_id == tenant_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + .count() + ), } @@ -227,7 +313,11 @@ DOCUMENTATION = ( "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." + "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"), @@ -242,8 +332,7 @@ DOCUMENTATION = ( DocumentationLink( label="Repository domain boundary", href=( - "govoplan-risk-compliance/" - "docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md" + "govoplan-risk-compliance/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md" ), kind="repository", ), @@ -260,6 +349,20 @@ DOCUMENTATION = ( "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." + ), }, ), ) @@ -292,10 +395,14 @@ manifest = ModuleManifest( role_templates=ROLE_TEMPLATES, route_factory=_route_factory, capability_factories={ - CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING: ( - _sanctions_screening_provider - ), + 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", @@ -303,7 +410,7 @@ manifest = ModuleManifest( FrontendRoute( path="/risk-compliance", component="RiskCompliancePage", - required_any=(SANCTIONS_READ_SCOPE,), + required_any=(READ_SCOPE, SANCTIONS_READ_SCOPE), order=115, surface_id="risk_compliance.workspace", ), @@ -313,7 +420,7 @@ manifest = ModuleManifest( path="/risk-compliance", label="Risk Compliance", icon="shield-check", - required_any=(SANCTIONS_READ_SCOPE,), + required_any=(READ_SCOPE, SANCTIONS_READ_SCOPE), order=115, surface_id="risk_compliance.navigation", ), @@ -340,15 +447,21 @@ manifest = ModuleManifest( label="Sanctions review queue", order=40, ), + ViewSurface( + id="risk_compliance.assurance.graph", + module_id=MODULE_ID, + kind="section", + label="Assurance graph", + order=50, + ), ), ), 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" - ), + script_location=str(Path(__file__).with_name("migrations") / "versions"), migration_after=("connectors",), retirement_supported=True, retirement_provider=drop_table_retirement_provider( @@ -356,8 +469,9 @@ manifest = ModuleManifest( label="Risk Compliance", ), retirement_notes=( - "Destructive retirement removes immutable sanctions list, " - "screening, and review evidence after a database snapshot." + "Destructive retirement removes immutable assurance graph, " + "sanctions list, screening, and review evidence after a database " + "snapshot." ), ), uninstall_guard_providers=( diff --git a/src/govoplan_risk_compliance/backend/migrations/versions/b9c0d1e2f3a4_v0114_assurance_graph.py b/src/govoplan_risk_compliance/backend/migrations/versions/b9c0d1e2f3a4_v0114_assurance_graph.py new file mode 100644 index 0000000..5a57544 --- /dev/null +++ b/src/govoplan_risk_compliance/backend/migrations/versions/b9c0d1e2f3a4_v0114_assurance_graph.py @@ -0,0 +1,180 @@ +"""Add the effective-dated horizontal assurance graph. + +Revision ID: b9c0d1e2f3a4 +Revises: a8b9c0d1e2f3 +Create Date: 2026-08-01 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "b9c0d1e2f3a4" +down_revision = "a8b9c0d1e2f3" +branch_labels = None +depends_on = None + + +def _timestamps() -> tuple[sa.Column, sa.Column]: + return ( + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def upgrade() -> None: + op.create_table( + "risk_assurance_nodes", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("stable_id", sa.String(length=255), nullable=False), + sa.Column("kind", sa.String(length=40), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("previous_revision_id", sa.String(length=36), nullable=True), + sa.Column("label", sa.String(length=500), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("state", sa.String(length=40), nullable=False), + sa.Column("owner_ref", sa.String(length=500), nullable=False), + sa.Column("scope_ref", sa.String(length=500), nullable=True), + sa.Column("governed_object_ref", sa.String(length=1000), nullable=True), + sa.Column("valid_from", sa.DateTime(timezone=True), nullable=False), + sa.Column("valid_to", sa.DateTime(timezone=True), nullable=True), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("superseded_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("legal_basis_refs", sa.JSON(), nullable=False), + sa.Column("policy_refs", sa.JSON(), nullable=False), + sa.Column("evidence_refs", sa.JSON(), nullable=False), + sa.Column("classification", sa.String(length=50), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + *_timestamps(), + sa.ForeignKeyConstraint( + ["previous_revision_id"], + ["risk_assurance_nodes.id"], + name=op.f( + "fk_risk_assurance_nodes_previous_revision_id_" + "risk_assurance_nodes" + ), + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_risk_assurance_nodes")), + sa.UniqueConstraint( + "tenant_id", + "stable_id", + "revision", + name="uq_risk_assurance_node_revision", + ), + ) + _indexes( + "risk_assurance_nodes", + ( + "tenant_id", + "stable_id", + "kind", + "state", + "owner_ref", + "scope_ref", + "governed_object_ref", + "valid_from", + "valid_to", + "recorded_at", + "superseded_at", + "classification", + "created_by", + ), + ) + op.create_index( + "ix_risk_assurance_node_current", + "risk_assurance_nodes", + ["tenant_id", "kind", "state", "superseded_at"], + ) + op.create_index( + "ix_risk_assurance_node_governed_object", + "risk_assurance_nodes", + ["tenant_id", "governed_object_ref", "superseded_at"], + ) + + op.create_table( + "risk_assurance_edges", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("stable_id", sa.String(length=255), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("previous_revision_id", sa.String(length=36), nullable=True), + sa.Column("source_node_ref", sa.String(length=255), nullable=False), + sa.Column("target_node_ref", sa.String(length=255), nullable=False), + sa.Column("relation", sa.String(length=50), nullable=False), + sa.Column("state", sa.String(length=30), nullable=False), + sa.Column("owner_ref", sa.String(length=500), nullable=False), + sa.Column("scope_ref", sa.String(length=500), nullable=True), + sa.Column("valid_from", sa.DateTime(timezone=True), nullable=False), + sa.Column("valid_to", sa.DateTime(timezone=True), nullable=True), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("superseded_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("legal_basis_refs", sa.JSON(), nullable=False), + sa.Column("policy_refs", sa.JSON(), nullable=False), + sa.Column("evidence_refs", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + *_timestamps(), + sa.ForeignKeyConstraint( + ["previous_revision_id"], + ["risk_assurance_edges.id"], + name=op.f( + "fk_risk_assurance_edges_previous_revision_id_" + "risk_assurance_edges" + ), + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_risk_assurance_edges")), + sa.UniqueConstraint( + "tenant_id", + "stable_id", + "revision", + name="uq_risk_assurance_edge_revision", + ), + ) + _indexes( + "risk_assurance_edges", + ( + "tenant_id", + "stable_id", + "source_node_ref", + "target_node_ref", + "relation", + "state", + "owner_ref", + "scope_ref", + "valid_from", + "valid_to", + "recorded_at", + "superseded_at", + "created_by", + ), + ) + op.create_index( + "ix_risk_assurance_edge_source", + "risk_assurance_edges", + ["tenant_id", "source_node_ref", "state", "superseded_at"], + ) + op.create_index( + "ix_risk_assurance_edge_target", + "risk_assurance_edges", + ["tenant_id", "target_node_ref", "state", "superseded_at"], + ) + + +def downgrade() -> None: + op.drop_table("risk_assurance_edges") + op.drop_table("risk_assurance_nodes") + + +def _indexes(table_name: str, columns: tuple[str, ...]) -> None: + for column in columns: + op.create_index( + op.f(f"ix_{table_name}_{column}"), + table_name, + [column], + ) diff --git a/src/govoplan_risk_compliance/backend/router.py b/src/govoplan_risk_compliance/backend/router.py index 3f523fd..9e8eac0 100644 --- a/src/govoplan_risk_compliance/backend/router.py +++ b/src/govoplan_risk_compliance/backend/router.py @@ -7,6 +7,23 @@ from govoplan_core.audit.logging import audit_from_principal from govoplan_core.auth import ApiPrincipal, get_api_principal from govoplan_core.core.sanctions import sanctions_snapshot_provider from govoplan_core.db.session import get_session +from govoplan_risk_compliance.backend.assurance import ( + AssuranceEdgeInput, + AssuranceNodeInput, + RiskAssuranceAccessError, + RiskAssuranceConflictError, + RiskAssuranceError, + RiskAssuranceNotFoundError, + assurance_graph, + assurance_summary, + create_assurance_edge, + create_assurance_node, + list_assurance_edge_history, + list_assurance_node_history, + list_assurance_nodes, + revise_assurance_edge, + revise_assurance_node, +) from govoplan_risk_compliance.backend.review import ( DispositionInput, get_candidate, @@ -23,6 +40,16 @@ from govoplan_risk_compliance.backend.sanctions_catalog import ( list_list_snapshots, ) from govoplan_risk_compliance.backend.schemas import ( + AssuranceEdgeListResponse, + AssuranceEdgeResponse, + AssuranceEdgeRevisionRequest, + AssuranceEdgeWrite, + AssuranceGraphResponse, + AssuranceNodeListResponse, + AssuranceNodeResponse, + AssuranceNodeRevisionRequest, + AssuranceNodeWrite, + AssuranceSummaryResponse, CandidateDetailResponse, CandidateResponse, ConnectorSnapshotListResponse, @@ -92,6 +119,292 @@ def _http_error(exc: RiskSanctionsError) -> HTTPException: ) +def _assurance_http_error(exc: RiskAssuranceError) -> HTTPException: + if isinstance(exc, RiskAssuranceNotFoundError): + return HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) + if isinstance(exc, RiskAssuranceAccessError): + return HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=str(exc), + ) + if isinstance(exc, RiskAssuranceConflictError): + return HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=str(exc), + ) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) + + +@router.get( + "/assurance/summary", + response_model=AssuranceSummaryResponse, +) +def api_assurance_summary( + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceSummaryResponse: + try: + payload = assurance_summary(session, principal) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + return AssuranceSummaryResponse.model_validate(payload) + + +@router.get( + "/assurance/nodes", + response_model=AssuranceNodeListResponse, +) +def api_list_assurance_nodes( + kind: str | None = Query(default=None, max_length=40), + state: str | None = Query(default=None, max_length=40), + governed_object_ref: str | None = Query(default=None, max_length=1000), + query: str | None = Query(default=None, max_length=500), + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceNodeListResponse: + try: + items = list_assurance_nodes( + session, + principal, + kind=kind, + state=state, + governed_object_ref=governed_object_ref, + query=query, + limit=limit, + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + return AssuranceNodeListResponse( + nodes=[_assurance_node_response(item) for item in items] + ) + + +@router.post( + "/assurance/nodes", + response_model=AssuranceNodeResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_assurance_node( + payload: AssuranceNodeWrite, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceNodeResponse: + try: + item = create_assurance_node( + session, + principal, + value=_assurance_node_input(payload), + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + audit_from_principal( + session, + principal, + action="risk_compliance.assurance_node.created", + scope="tenant", + object_type="risk_assurance_node", + object_id=item.stable_id, + details={ + "kind": item.kind, + "state": item.state, + "revision": item.revision, + }, + ) + session.commit() + return _assurance_node_response(item) + + +@router.post( + "/assurance/nodes/revise", + response_model=AssuranceNodeResponse, +) +def api_revise_assurance_node( + payload: AssuranceNodeRevisionRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceNodeResponse: + try: + item = revise_assurance_node( + session, + principal, + stable_id=payload.stable_id, + expected_revision=payload.expected_revision, + value=_assurance_node_input(payload), + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + audit_from_principal( + session, + principal, + action="risk_compliance.assurance_node.revised", + scope="tenant", + object_type="risk_assurance_node", + object_id=item.stable_id, + details={ + "kind": item.kind, + "state": item.state, + "revision": item.revision, + }, + ) + session.commit() + return _assurance_node_response(item) + + +@router.get( + "/assurance/node-history", + response_model=AssuranceNodeListResponse, +) +def api_assurance_node_history( + stable_id: str = Query(min_length=1, max_length=255), + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceNodeListResponse: + try: + items = list_assurance_node_history( + session, + principal, + stable_id=stable_id, + limit=limit, + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + return AssuranceNodeListResponse( + nodes=[_assurance_node_response(item) for item in items] + ) + + +@router.post( + "/assurance/edges", + response_model=AssuranceEdgeResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_assurance_edge( + payload: AssuranceEdgeWrite, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceEdgeResponse: + try: + item = create_assurance_edge( + session, + principal, + value=_assurance_edge_input(payload), + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + audit_from_principal( + session, + principal, + action="risk_compliance.assurance_edge.created", + scope="tenant", + object_type="risk_assurance_edge", + object_id=item.stable_id, + details={ + "relation": item.relation, + "state": item.state, + "revision": item.revision, + }, + ) + session.commit() + return _assurance_edge_response(item) + + +@router.post( + "/assurance/edges/revise", + response_model=AssuranceEdgeResponse, +) +def api_revise_assurance_edge( + payload: AssuranceEdgeRevisionRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceEdgeResponse: + try: + item = revise_assurance_edge( + session, + principal, + stable_id=payload.stable_id, + expected_revision=payload.expected_revision, + value=_assurance_edge_input(payload), + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + audit_from_principal( + session, + principal, + action="risk_compliance.assurance_edge.revised", + scope="tenant", + object_type="risk_assurance_edge", + object_id=item.stable_id, + details={ + "relation": item.relation, + "state": item.state, + "revision": item.revision, + }, + ) + session.commit() + return _assurance_edge_response(item) + + +@router.get( + "/assurance/edge-history", + response_model=AssuranceEdgeListResponse, +) +def api_assurance_edge_history( + stable_id: str = Query(min_length=1, max_length=255), + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceEdgeListResponse: + try: + items = list_assurance_edge_history( + session, + principal, + stable_id=stable_id, + limit=limit, + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + return AssuranceEdgeListResponse( + edges=[_assurance_edge_response(item) for item in items] + ) + + +@router.get( + "/assurance/graph", + response_model=AssuranceGraphResponse, +) +def api_assurance_graph( + root_ref: str = Query(min_length=1, max_length=255), + max_depth: int = Query(default=4, ge=0, le=8), + limit: int = Query(default=200, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> AssuranceGraphResponse: + try: + graph = assurance_graph( + session, + principal, + root_ref=root_ref, + max_depth=max_depth, + limit=limit, + ) + except RiskAssuranceError as exc: + raise _assurance_http_error(exc) from exc + return AssuranceGraphResponse( + root_ref=graph.root_ref, + nodes=[_assurance_node_response(item) for item in graph.nodes], + edges=[_assurance_edge_response(item) for item in graph.edges], + truncated=graph.truncated, + ) + + @router.get( "/sanctions/source-snapshots", response_model=ConnectorSnapshotListResponse, @@ -689,4 +1002,56 @@ def _disposition_response(item) -> DispositionResponse: ) +def _assurance_node_input(item: AssuranceNodeWrite) -> AssuranceNodeInput: + return AssuranceNodeInput( + stable_id=item.stable_id, + kind=item.kind, + label=item.label, + description=item.description, + state=item.state, + owner_ref=item.owner_ref, + scope_ref=item.scope_ref, + governed_object_ref=item.governed_object_ref, + valid_from=item.valid_from, + valid_to=item.valid_to, + provenance=item.provenance, + legal_basis_refs=tuple(item.legal_basis_refs), + policy_refs=tuple(item.policy_refs), + evidence_refs=tuple(item.evidence_refs), + classification=item.classification, + ) + + +def _assurance_edge_input(item: AssuranceEdgeWrite) -> AssuranceEdgeInput: + return AssuranceEdgeInput( + stable_id=item.stable_id, + source_node_ref=item.source_node_ref, + target_node_ref=item.target_node_ref, + relation=item.relation, + state=item.state, + owner_ref=item.owner_ref, + scope_ref=item.scope_ref, + valid_from=item.valid_from, + valid_to=item.valid_to, + provenance=item.provenance, + legal_basis_refs=tuple(item.legal_basis_refs), + policy_refs=tuple(item.policy_refs), + evidence_refs=tuple(item.evidence_refs), + ) + + +def _assurance_node_response(item) -> AssuranceNodeResponse: + return AssuranceNodeResponse.model_validate( + item, + from_attributes=True, + ) + + +def _assurance_edge_response(item) -> AssuranceEdgeResponse: + return AssuranceEdgeResponse.model_validate( + item, + from_attributes=True, + ) + + __all__ = ["router"] diff --git a/src/govoplan_risk_compliance/backend/schemas.py b/src/govoplan_risk_compliance/backend/schemas.py index 1950907..0f4e1b7 100644 --- a/src/govoplan_risk_compliance/backend/schemas.py +++ b/src/govoplan_risk_compliance/backend/schemas.py @@ -352,7 +352,124 @@ class DispositionListResponse(BaseModel): dispositions: list[DispositionResponse] +AssuranceNodeKind = Literal[ + "obligation", + "governed_object", + "risk", + "control", + "evidence", + "finding", + "corrective_measure", + "effectiveness_review", +] +AssuranceEdgeRelation = Literal[ + "applies_to", + "exposes_risk", + "mitigated_by", + "evidenced_by", + "results_in", + "addressed_by", + "reviewed_by", +] + + +class AssuranceNodeWrite(BaseModel): + stable_id: str = Field(min_length=1, max_length=255) + kind: AssuranceNodeKind + label: str = Field(min_length=1, max_length=500) + description: str | None = Field(default=None, max_length=20_000) + state: str = Field(min_length=1, max_length=40) + owner_ref: str = Field(min_length=1, max_length=500) + scope_ref: str | None = Field(default=None, max_length=500) + governed_object_ref: str | None = Field(default=None, max_length=1000) + valid_from: datetime + valid_to: datetime | None = None + provenance: dict[str, Any] = Field(default_factory=dict) + legal_basis_refs: list[str] = Field(default_factory=list, max_length=100) + policy_refs: list[str] = Field(default_factory=list, max_length=100) + evidence_refs: list[str] = Field(default_factory=list, max_length=100) + classification: str = Field(default="internal", min_length=1, max_length=50) + + +class AssuranceNodeRevisionRequest(AssuranceNodeWrite): + expected_revision: int = Field(ge=1) + + +class AssuranceNodeResponse(AssuranceNodeWrite): + id: str + revision: int + previous_revision_id: str | None + recorded_at: datetime + superseded_at: datetime | None + created_by: str | None + created_at: datetime + updated_at: datetime + + +class AssuranceNodeListResponse(BaseModel): + nodes: list[AssuranceNodeResponse] + + +class AssuranceEdgeWrite(BaseModel): + stable_id: str = Field(min_length=1, max_length=255) + source_node_ref: str = Field(min_length=1, max_length=255) + target_node_ref: str = Field(min_length=1, max_length=255) + relation: AssuranceEdgeRelation + state: Literal["active", "suspended", "retired"] = "active" + owner_ref: str = Field(min_length=1, max_length=500) + scope_ref: str | None = Field(default=None, max_length=500) + valid_from: datetime + valid_to: datetime | None = None + provenance: dict[str, Any] = Field(default_factory=dict) + legal_basis_refs: list[str] = Field(default_factory=list, max_length=100) + policy_refs: list[str] = Field(default_factory=list, max_length=100) + evidence_refs: list[str] = Field(default_factory=list, max_length=100) + + +class AssuranceEdgeRevisionRequest(AssuranceEdgeWrite): + expected_revision: int = Field(ge=1) + + +class AssuranceEdgeResponse(AssuranceEdgeWrite): + id: str + revision: int + previous_revision_id: str | None + recorded_at: datetime + superseded_at: datetime | None + created_by: str | None + created_at: datetime + updated_at: datetime + + +class AssuranceEdgeListResponse(BaseModel): + edges: list[AssuranceEdgeResponse] + + +class AssuranceGraphResponse(BaseModel): + root_ref: str + nodes: list[AssuranceNodeResponse] + edges: list[AssuranceEdgeResponse] + truncated: bool + + +class AssuranceSummaryResponse(BaseModel): + node_count: int + edge_count: int + by_kind: dict[str, int] + by_state: dict[str, int] + + __all__ = [ + "AssuranceEdgeListResponse", + "AssuranceEdgeResponse", + "AssuranceEdgeRevisionRequest", + "AssuranceEdgeWrite", + "AssuranceGraphResponse", + "AssuranceNodeListResponse", + "AssuranceNodeResponse", + "AssuranceNodeRevisionRequest", + "AssuranceNodeWrite", + "AssuranceSummaryResponse", "CandidateDetailResponse", "CandidateResponse", "ConnectorSnapshotListResponse", diff --git a/src/govoplan_risk_compliance/backend/screening.py b/src/govoplan_risk_compliance/backend/screening.py index 1801e55..e322fc2 100644 --- a/src/govoplan_risk_compliance/backend/screening.py +++ b/src/govoplan_risk_compliance/backend/screening.py @@ -152,11 +152,12 @@ def run_screening( raise RiskSanctionsConflictError( "Idempotency key was already used for another screening request." ) - return get_screening_run( + return _screening_result( session, principal, run_id=existing.id, - ), False + created=False, + ) list_snapshot = _visible_list_snapshot( session, @@ -207,11 +208,12 @@ def run_screening( run.outcome = list_state run.completed_at = utcnow() session.flush() - return get_screening_run( + return _screening_result( session, principal, run_id=run.id, - ), True + created=True, + ) entries = tuple( session.scalars( @@ -235,11 +237,12 @@ def run_screening( } run.completed_at = utcnow() session.flush() - return get_screening_run( + return _screening_result( session, principal, run_id=run.id, - ), True + created=True, + ) candidates = sorted( ( @@ -297,11 +300,12 @@ def run_screening( run.outcome = "potential" if candidates else "clear" run.completed_at = utcnow() session.flush() - return get_screening_run( + return _screening_result( session, principal, run_id=run.id, - ), True + created=True, + ) def screening_evidence_ref(run_id: str) -> str: @@ -313,6 +317,31 @@ def screening_evidence_ref(run_id: str) -> str: return f"{SCREENING_EVIDENCE_PREFIX}{clean_run_id}" +def _screening_result( + session: Session, + principal: ApiPrincipal, + *, + run_id: str, + created: bool, +) -> tuple[RiskScreeningRun, bool]: + run = get_screening_run( + session, + principal, + run_id=run_id, + ) + if run.status == "completed": + from govoplan_risk_compliance.backend.assurance import ( + record_sanctions_assurance, + ) + + record_sanctions_assurance( + session, + principal, + run=run, + ) + return run, created + + def assess_screening_freshness( session: Session, principal: ApiPrincipal, diff --git a/src/govoplan_risk_compliance/backend/search_source.py b/src/govoplan_risk_compliance/backend/search_source.py new file mode 100644 index 0000000..046275b --- /dev/null +++ b/src/govoplan_risk_compliance/backend/search_source.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from urllib.parse import quote + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.modules import ModuleContext +from govoplan_core.core.search import ( + SearchAuthorizationRequest, + SearchBackfillPage, + SearchBackfillRequest, + SearchDocument, + SearchResourceType, +) +from govoplan_risk_compliance.backend.db.models import RiskAssuranceNode +from govoplan_risk_compliance.backend.permissions import ( + ADMIN_SCOPE, + READ_SCOPE, + WRITE_SCOPE, +) + + +PROVIDER_ID = "risk_compliance.assurance" +RESOURCE_TYPE = "risk_assurance_node" + + +class RiskAssuranceSearchSource: + def resource_types(self) -> Sequence[SearchResourceType]: + return ( + SearchResourceType( + provider_id=PROVIDER_ID, + module_id="risk_compliance", + resource_type=RESOURCE_TYPE, + label="Assurance graph", + requires_authorization_recheck=True, + ), + ) + + def backfill( + self, + session: object, + *, + request: SearchBackfillRequest, + ) -> SearchBackfillPage: + db = _session(session) + if ( + request.provider_id != PROVIDER_ID + or request.resource_type != RESOURCE_TYPE + ): + raise ValueError("Unsupported Risk Compliance search source.") + statement = select(RiskAssuranceNode).where( + RiskAssuranceNode.tenant_id == request.tenant_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + if request.cursor: + statement = statement.where( + RiskAssuranceNode.stable_id > request.cursor + ) + rows = list( + db.scalars( + statement.order_by(RiskAssuranceNode.stable_id).limit( + request.limit + 1 + ) + ) + ) + has_more = len(rows) > request.limit + selected = rows[: request.limit] + high_watermark = db.scalar( + select(func.max(RiskAssuranceNode.recorded_at)).where( + RiskAssuranceNode.tenant_id == request.tenant_id, + RiskAssuranceNode.superseded_at.is_(None), + ) + ) + return SearchBackfillPage( + documents=tuple(_document(item) for item in selected), + next_cursor=( + selected[-1].stable_id + if has_more and selected + else None + ), + complete=not has_more, + high_watermark=( + high_watermark.isoformat() + if high_watermark is not None + else None + ), + ) + + def authorize( + self, + session: object, + principal: object, + *, + requests: Sequence[SearchAuthorizationRequest], + ) -> Mapping[str, bool]: + decisions = {request.reference.key: False for request in requests} + if not isinstance(principal, ApiPrincipal) or not any( + principal.has(scope) + for scope in (READ_SCOPE, WRITE_SCOPE, ADMIN_SCOPE) + ): + return decisions + valid = tuple( + request + for request in requests + if request.reference.tenant_id == principal.tenant_id + and request.reference.module_id == "risk_compliance" + and request.reference.resource_type == RESOURCE_TYPE + ) + if not valid: + return decisions + ids = {request.reference.resource_id for request in valid} + existing = set( + _session(session).scalars( + select(RiskAssuranceNode.stable_id).where( + RiskAssuranceNode.tenant_id == principal.tenant_id, + RiskAssuranceNode.stable_id.in_(ids), + RiskAssuranceNode.superseded_at.is_(None), + ) + ) + ) + for request in valid: + decisions[request.reference.key] = ( + request.reference.resource_id in existing + ) + return decisions + + +def create_risk_assurance_search_source( + _context: ModuleContext, +) -> RiskAssuranceSearchSource: + return RiskAssuranceSearchSource() + + +def _document(item: RiskAssuranceNode) -> SearchDocument: + return SearchDocument( + tenant_id=item.tenant_id, + module_id="risk_compliance", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + resource_id=item.stable_id, + title=item.label, + url=( + "/risk-compliance?view=assurance&node=" + f"{quote(item.stable_id, safe='')}" + ), + summary=( + item.description + if item.classification in {"public", "internal"} + else None + ), + keywords=(item.kind, item.state, item.classification), + visibility="tenant", + source_revision=str(item.revision), + source_updated_at=item.recorded_at, + metadata={ + "kind": item.kind, + "state": item.state, + "classification": item.classification, + "revision": item.revision, + }, + requires_authorization_recheck=True, + ) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError( + "Risk assurance search requires a SQLAlchemy session." + ) + return value + + +__all__ = [ + "PROVIDER_ID", + "RESOURCE_TYPE", + "RiskAssuranceSearchSource", + "create_risk_assurance_search_source", +] diff --git a/tests/test_assurance_graph.py b/tests/test_assurance_graph.py new file mode 100644 index 0000000..77e5ac8 --- /dev/null +++ b/tests/test_assurance_graph.py @@ -0,0 +1,378 @@ +from __future__ import annotations + +from datetime import UTC, datetime +from types import SimpleNamespace +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.search import ( + SearchAuthorizationRequest, + SearchBackfillRequest, + SearchResourceReference, +) +from govoplan_core.db.base import Base +from govoplan_risk_compliance.backend.assurance import ( + AssuranceEdgeInput, + AssuranceNodeInput, + RiskAssuranceAccessError, + RiskAssuranceConflictError, + RiskAssuranceError, + RiskAssuranceNotFoundError, + assurance_graph, + assurance_summary, + create_assurance_edge, + create_assurance_node, + list_assurance_edge_history, + list_assurance_node_history, + list_assurance_nodes, + revise_assurance_edge, + revise_assurance_node, +) +from govoplan_risk_compliance.backend.db.models import ( + RiskAssuranceEdge, + RiskAssuranceNode, +) +from govoplan_risk_compliance.backend.permissions import ( + READ_SCOPE, + WRITE_SCOPE, +) +from govoplan_risk_compliance.backend.search_source import ( + PROVIDER_ID, + RESOURCE_TYPE, + RiskAssuranceSearchSource, +) + + +NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC) +TABLES = ( + RiskAssuranceNode.__table__, + RiskAssuranceEdge.__table__, +) + + +def principal( + tenant_id: str = "tenant-1", + *, + scopes: tuple[str, ...] = (READ_SCOPE, WRITE_SCOPE), +) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id=f"account-{tenant_id}", + membership_id=f"membership-{tenant_id}", + tenant_id=tenant_id, + scopes=frozenset(scopes), + ), + account=SimpleNamespace(id=f"account-{tenant_id}"), + user=SimpleNamespace(id=f"account-{tenant_id}"), + ) + + +class AssuranceGraphTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(self.engine, tables=TABLES) + self.session = Session(self.engine) + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_non_sanctions_control_uses_complete_revisioned_graph(self) -> None: + node_specs = ( + ("retention-obligation", "obligation", "active"), + ("customer-register", "governed_object", "active"), + ("over-retention-risk", "risk", "identified"), + ("retention-control", "control", "implemented"), + ("retention-evidence", "evidence", "current"), + ("retention-finding", "finding", "open"), + ("retention-measure", "corrective_measure", "planned"), + ("retention-review", "effectiveness_review", "pending"), + ) + nodes = {} + for stable_id, kind, state in node_specs: + nodes[stable_id] = create_assurance_node( + self.session, + principal(), + value=AssuranceNodeInput( + stable_id=stable_id, + kind=kind, + label=stable_id.replace("-", " ").title(), + state=state, + owner_ref="organizations:function:data-governance", + scope_ref="datasource:customer-register", + governed_object_ref=( + "datasources:customer-register" + if kind == "governed_object" + else None + ), + valid_from=NOW, + provenance={"fixture": "non-sanctions"}, + legal_basis_refs=("law:retention:2026",), + ), + ) + relations = ( + ("retention-obligation", "applies_to", "customer-register"), + ("customer-register", "exposes_risk", "over-retention-risk"), + ("over-retention-risk", "mitigated_by", "retention-control"), + ("retention-control", "evidenced_by", "retention-evidence"), + ("retention-evidence", "results_in", "retention-finding"), + ("retention-finding", "addressed_by", "retention-measure"), + ("retention-measure", "reviewed_by", "retention-review"), + ) + edges = [] + for index, (source, relation, target) in enumerate(relations, start=1): + edges.append( + create_assurance_edge( + self.session, + principal(), + value=AssuranceEdgeInput( + stable_id=f"retention-edge-{index}", + source_node_ref=source, + target_node_ref=target, + relation=relation, + owner_ref="organizations:function:data-governance", + valid_from=NOW, + provenance={"fixture": "non-sanctions"}, + ), + ) + ) + self.session.commit() + + graph = assurance_graph( + self.session, + principal(), + root_ref="retention-obligation", + max_depth=8, + limit=100, + ) + summary = assurance_summary(self.session, principal()) + + self.assertEqual(8, len(graph.nodes)) + self.assertEqual(7, len(graph.edges)) + self.assertFalse(graph.truncated) + self.assertEqual(8, summary["node_count"]) + self.assertEqual(7, summary["edge_count"]) + self.assertEqual(1, summary["by_kind"]["effectiveness_review"]) + self.assertEqual("control", nodes["retention-control"].kind) + self.assertEqual("reviewed_by", edges[-1].relation) + + def test_node_and_edge_revisions_are_immutable_and_occ_guarded(self) -> None: + risk = self._node("risk-1", "risk", "identified") + control = self._node("control-1", "control", "implemented") + edge = create_assurance_edge( + self.session, + principal(), + value=AssuranceEdgeInput( + stable_id="edge-1", + source_node_ref=risk.stable_id, + target_node_ref=control.stable_id, + relation="mitigated_by", + owner_ref="function:risk-owner", + valid_from=NOW, + ), + ) + revised = revise_assurance_node( + self.session, + principal(), + stable_id=risk.stable_id, + expected_revision=1, + value=AssuranceNodeInput( + stable_id=risk.stable_id, + kind="risk", + label="Risk 1", + state="assessed", + owner_ref="function:risk-owner", + valid_from=NOW, + evidence_refs=("evidence:assessment-1",), + ), + ) + revised_edge = revise_assurance_edge( + self.session, + principal(), + stable_id=edge.stable_id, + expected_revision=1, + value=AssuranceEdgeInput( + stable_id=edge.stable_id, + source_node_ref=risk.stable_id, + target_node_ref=control.stable_id, + relation="mitigated_by", + state="suspended", + owner_ref="function:risk-owner", + valid_from=NOW, + evidence_refs=("evidence:suspension-1",), + ), + ) + self.session.commit() + + self.assertEqual(2, revised.revision) + self.assertEqual(2, revised_edge.revision) + self.assertEqual( + [2, 1], + [ + item.revision + for item in list_assurance_node_history( + self.session, + principal(), + stable_id="risk-1", + ) + ], + ) + self.assertEqual( + [2, 1], + [ + item.revision + for item in list_assurance_edge_history( + self.session, + principal(), + stable_id="edge-1", + ) + ], + ) + with self.assertRaisesRegex( + RiskAssuranceConflictError, + "current revision is 2", + ): + revise_assurance_node( + self.session, + principal(), + stable_id="risk-1", + expected_revision=1, + value=AssuranceNodeInput( + stable_id="risk-1", + kind="risk", + label="Risk 1", + state="closed", + owner_ref="function:risk-owner", + valid_from=NOW, + ), + ) + + def test_access_and_tenant_boundaries_are_enforced(self) -> None: + self._node("risk-1", "risk", "identified") + self.session.commit() + + self.assertEqual( + (), + list_assurance_nodes(self.session, principal("tenant-2")), + ) + with self.assertRaises(RiskAssuranceNotFoundError): + assurance_graph( + self.session, + principal("tenant-2"), + root_ref="risk-1", + ) + with self.assertRaises(RiskAssuranceAccessError): + list_assurance_nodes( + self.session, + principal(scopes=()), + ) + + def test_programmatic_inputs_enforce_text_and_provenance_bounds(self) -> None: + with self.assertRaisesRegex(RiskAssuranceError, "description is limited"): + AssuranceNodeInput( + stable_id="risk-oversized-description", + kind="risk", + label="Oversized risk", + state="identified", + owner_ref="function:risk-owner", + valid_from=NOW, + description="x" * 20_001, + ) + with self.assertRaisesRegex(RiskAssuranceError, "provenance is limited"): + AssuranceNodeInput( + stable_id="risk-oversized-provenance", + kind="risk", + label="Oversized provenance", + state="identified", + owner_ref="function:risk-owner", + valid_from=NOW, + provenance={"payload": "x" * 100_001}, + ) + + def test_relation_shape_is_validated(self) -> None: + self._node("risk-1", "risk", "identified") + self._node("evidence-1", "evidence", "current") + + with self.assertRaisesRegex( + RiskAssuranceConflictError, + "requires risk -> control", + ): + create_assurance_edge( + self.session, + principal(), + value=AssuranceEdgeInput( + stable_id="invalid-edge", + source_node_ref="risk-1", + target_node_ref="evidence-1", + relation="mitigated_by", + owner_ref="function:risk-owner", + valid_from=NOW, + ), + ) + + def test_search_backfill_and_authorization_are_tenant_safe(self) -> None: + self._node("risk-1", "risk", "identified") + self.session.commit() + provider = RiskAssuranceSearchSource() + + page = provider.backfill( + self.session, + request=SearchBackfillRequest( + tenant_id="tenant-1", + provider_id=PROVIDER_ID, + resource_type=RESOURCE_TYPE, + rebuild_id="rebuild-1", + ), + ) + request = SearchAuthorizationRequest( + reference=SearchResourceReference( + tenant_id="tenant-1", + module_id="risk_compliance", + resource_type=RESOURCE_TYPE, + resource_id="risk-1", + ), + source_revision="1", + ) + + self.assertEqual(1, len(page.documents)) + self.assertTrue( + provider.authorize( + self.session, + principal(), + requests=(request,), + )[request.reference.key] + ) + self.assertFalse( + provider.authorize( + self.session, + principal("tenant-2"), + requests=(request,), + )[request.reference.key] + ) + + def _node( + self, + stable_id: str, + kind: str, + state: str, + ) -> RiskAssuranceNode: + return create_assurance_node( + self.session, + principal(), + value=AssuranceNodeInput( + stable_id=stable_id, + kind=kind, + label=stable_id.replace("-", " ").title(), + state=state, + owner_ref="function:risk-owner", + valid_from=NOW, + ), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 1184401..e7f0667 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -48,6 +48,11 @@ class ManifestTests(unittest.TestCase): self.assertIsNotNone(manifest.route_factory) self.assertIsNotNone(manifest.migration_spec) self.assertIsNotNone(manifest.frontend) + self.assertEqual("vertical_slice", manifest.architecture.maturity) + self.assertIn( + "governance_overlay", + manifest.architecture.supported_authority_modes, + ) self.assertEqual( "connectors.sanctions_snapshots", manifest.requires_interfaces[0].name, diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 29941da..4574398 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -26,7 +26,7 @@ class RiskComplianceMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "a8b9c0d1e2f3", + "b9c0d1e2f3a4", set( MigrationContext.configure( connection @@ -49,6 +49,8 @@ class RiskComplianceMigrationTests(unittest.TestCase): "risk_screening_exceptions", tables, ) + self.assertIn("risk_assurance_nodes", tables) + self.assertIn("risk_assurance_edges", tables) finally: engine.dispose() diff --git a/tests/test_sanctions_screening.py b/tests/test_sanctions_screening.py index 2d45e83..027175a 100644 --- a/tests/test_sanctions_screening.py +++ b/tests/test_sanctions_screening.py @@ -22,6 +22,8 @@ from govoplan_core.core.sanctions import ( ) from govoplan_core.db.base import Base, utcnow from govoplan_risk_compliance.backend.db.models import ( + RiskAssuranceEdge, + RiskAssuranceNode, RiskSanctionsAddress, RiskSanctionsAlias, RiskSanctionsDate, @@ -104,6 +106,8 @@ UN_XML = b""" """ TABLES = ( + RiskAssuranceNode.__table__, + RiskAssuranceEdge.__table__, RiskSanctionsListSnapshot.__table__, RiskSanctionsEntry.__table__, RiskSanctionsAlias.__table__, @@ -254,6 +258,27 @@ class SanctionsScreeningTests(unittest.TestCase): self.assertFalse( item.candidates[0].evidence[0]["auto_confirmed"] ) + self.assertEqual( + 6, + self.session.query(RiskAssuranceNode) + .filter( + RiskAssuranceNode.tenant_id == "tenant-1", + RiskAssuranceNode.superseded_at.is_(None), + ) + .count(), + ) + self.assertEqual( + 5, + self.session.query(RiskAssuranceEdge) + .filter( + RiskAssuranceEdge.tenant_id == "tenant-1", + RiskAssuranceEdge.superseded_at.is_(None), + ) + .count(), + ) + item_id = item.id + self.session.commit() + self.session.expire_all() replay, replay_created = run_screening( self.session, @@ -266,7 +291,9 @@ class SanctionsScreeningTests(unittest.TestCase): ), ) self.assertFalse(replay_created) - self.assertEqual(item.id, replay.id) + self.assertEqual(item_id, replay.id) + self.assertEqual(6, self.session.query(RiskAssuranceNode).count()) + self.assertEqual(5, self.session.query(RiskAssuranceEdge).count()) with self.assertRaises(RiskSanctionsConflictError): run_screening( diff --git a/webui/src/api/riskCompliance.ts b/webui/src/api/riskCompliance.ts index a8738ed..1c97d3b 100644 --- a/webui/src/api/riskCompliance.ts +++ b/webui/src/api/riskCompliance.ts @@ -145,6 +145,97 @@ export type CandidateDetail = { }; }; +export type AssuranceNodeKind = + | "obligation" + | "governed_object" + | "risk" + | "control" + | "evidence" + | "finding" + | "corrective_measure" + | "effectiveness_review"; + +export type AssuranceNode = { + id: string; + stable_id: string; + kind: AssuranceNodeKind; + revision: number; + previous_revision_id?: string | null; + label: string; + description?: string | null; + state: string; + owner_ref: string; + scope_ref?: string | null; + governed_object_ref?: string | null; + valid_from: string; + valid_to?: string | null; + recorded_at: string; + superseded_at?: string | null; + provenance: Record; + legal_basis_refs: string[]; + policy_refs: string[]; + evidence_refs: string[]; + classification: string; + created_by?: string | null; + created_at: string; + updated_at: string; +}; + +export type AssuranceEdge = { + id: string; + stable_id: string; + revision: number; + previous_revision_id?: string | null; + source_node_ref: string; + target_node_ref: string; + relation: string; + state: "active" | "suspended" | "retired"; + owner_ref: string; + scope_ref?: string | null; + valid_from: string; + valid_to?: string | null; + recorded_at: string; + superseded_at?: string | null; + provenance: Record; + legal_basis_refs: string[]; + policy_refs: string[]; + evidence_refs: string[]; + created_by?: string | null; + created_at: string; + updated_at: string; +}; + +export type AssuranceSummary = { + node_count: number; + edge_count: number; + by_kind: Record; + by_state: Record; +}; + +export type AssuranceNodeWrite = Omit< + AssuranceNode, + | "id" + | "revision" + | "previous_revision_id" + | "recorded_at" + | "superseded_at" + | "created_by" + | "created_at" + | "updated_at" +>; + +export type AssuranceEdgeWrite = Omit< + AssuranceEdge, + | "id" + | "revision" + | "previous_revision_id" + | "recorded_at" + | "superseded_at" + | "created_by" + | "created_at" + | "updated_at" +>; + export async function listConnectorSnapshots( settings: ApiSettings ) { @@ -251,3 +342,94 @@ export async function createDisposition( } ); } + +export async function listAssuranceNodes( + settings: ApiSettings, + filters: { + kind?: string; + state?: string; + query?: string; + governedObjectRef?: string; + } = {} +) { + return apiFetch<{ nodes: AssuranceNode[] }>( + settings, + apiPath("/api/v1/risk-compliance/assurance/nodes", { + kind: filters.kind, + state: filters.state, + query: filters.query, + governed_object_ref: filters.governedObjectRef, + limit: 500 + }) + ); +} + +export async function getAssuranceSummary(settings: ApiSettings) { + return apiFetch( + settings, + "/api/v1/risk-compliance/assurance/summary" + ); +} + +export async function getAssuranceGraph( + settings: ApiSettings, + rootRef: string +) { + return apiFetch<{ + root_ref: string; + nodes: AssuranceNode[]; + edges: AssuranceEdge[]; + truncated: boolean; + }>( + settings, + apiPath("/api/v1/risk-compliance/assurance/graph", { + root_ref: rootRef, + max_depth: 8, + limit: 500 + }) + ); +} + +export async function saveAssuranceNode( + settings: ApiSettings, + value: AssuranceNodeWrite, + expectedRevision?: number +) { + return apiFetch( + settings, + expectedRevision + ? "/api/v1/risk-compliance/assurance/nodes/revise" + : "/api/v1/risk-compliance/assurance/nodes", + { + method: "POST", + body: JSON.stringify({ + ...value, + ...(expectedRevision + ? { expected_revision: expectedRevision } + : {}) + }) + } + ); +} + +export async function saveAssuranceEdge( + settings: ApiSettings, + value: AssuranceEdgeWrite, + expectedRevision?: number +) { + return apiFetch( + settings, + expectedRevision + ? "/api/v1/risk-compliance/assurance/edges/revise" + : "/api/v1/risk-compliance/assurance/edges", + { + method: "POST", + body: JSON.stringify({ + ...value, + ...(expectedRevision + ? { expected_revision: expectedRevision } + : {}) + }) + } + ); +} diff --git a/webui/src/features/riskCompliance/RiskCompliancePage.tsx b/webui/src/features/riskCompliance/RiskCompliancePage.tsx index cd6961f..adca5b6 100644 --- a/webui/src/features/riskCompliance/RiskCompliancePage.tsx +++ b/webui/src/features/riskCompliance/RiskCompliancePage.tsx @@ -1,7 +1,10 @@ import { CheckCircle2, Database, + Network, + Pencil, Play, + Plus, RefreshCw, Scale, Upload @@ -13,6 +16,7 @@ import { useState, type FormEvent } from "react"; +import { useSearchParams } from "react-router"; import { Button, Dialog, @@ -28,12 +32,23 @@ import { } from "@govoplan/core-webui"; import { createDisposition, + getAssuranceGraph, + getAssuranceSummary, getCandidate, importListSnapshot, listConnectorSnapshots, + listAssuranceNodes, listListSnapshots, listReviewQueue, runScreening, + saveAssuranceEdge, + saveAssuranceNode, + type AssuranceEdge, + type AssuranceEdgeWrite, + type AssuranceNode, + type AssuranceNodeKind, + type AssuranceNodeWrite, + type AssuranceSummary, type CandidateDetail, type ConnectorSnapshot, type ListSnapshot, @@ -42,13 +57,18 @@ import { } from "../../api/riskCompliance"; -type ViewMode = "sources" | "screen" | "review"; +type ViewMode = "sources" | "screen" | "review" | "assurance"; export default function RiskCompliancePage({ settings, auth }: PlatformRouteContext) { - const [view, setView] = useState("review"); + const [searchParams] = useSearchParams(); + const requestedView = searchParams.get("view"); + const requestedAssuranceId = searchParams.get("node")?.trim() ?? ""; + const [view, setView] = useState(() => + requestedView === "assurance" ? "assurance" : "review" + ); const [sourceSnapshots, setSourceSnapshots] = useState< ConnectorSnapshot[] >([]); @@ -58,6 +78,17 @@ export default function RiskCompliancePage({ const [selectedCandidateId, setSelectedCandidateId] = useState(""); const [candidate, setCandidate] = useState(null); const [run, setRun] = useState(null); + const [assuranceNodes, setAssuranceNodes] = useState([]); + const [assuranceSummary, setAssuranceSummary] = + useState(null); + const [selectedAssuranceId, setSelectedAssuranceId] = useState( + requestedAssuranceId + ); + const [assuranceGraph, setAssuranceGraph] = useState<{ + nodes: AssuranceNode[]; + edges: AssuranceEdge[]; + truncated: boolean; + } | null>(null); const [loading, setLoading] = useState(true); const [busy, setBusy] = useState(false); const [error, setError] = useState(""); @@ -74,34 +105,71 @@ export default function RiskCompliancePage({ auth, "risk_compliance:sanctions:review" ); + const canReadAssurance = [ + "risk_compliance:workspace:read", + "risk_compliance:workspace:write", + "risk_compliance:workspace:admin" + ].some((scope) => hasScope(auth, scope)); + const canWriteAssurance = [ + "risk_compliance:workspace:write", + "risk_compliance:workspace:admin" + ].some((scope) => hasScope(auth, scope)); + + useEffect(() => { + if (requestedView === "assurance" && canReadAssurance) { + setView("assurance"); + if (requestedAssuranceId) { + setSelectedAssuranceId(requestedAssuranceId); + } + } + }, [canReadAssurance, requestedAssuranceId, requestedView]); + + useEffect(() => { + if (view === "review" && !canReview) { + setView(canReadAssurance ? "assurance" : "sources"); + } else if (view === "assurance" && !canReadAssurance) { + setView(canReview ? "review" : "sources"); + } + }, [canReadAssurance, canReview, view]); const refresh = useCallback(async () => { setLoading(true); setError(""); try { - const [sources, lists, reviewQueue] = await Promise.all([ + const [sources, lists, reviewQueue, assurance] = await Promise.all([ listConnectorSnapshots(settings), listListSnapshots(settings), canReview ? listReviewQueue(settings) - : Promise.resolve({ candidates: [] }) + : Promise.resolve({ candidates: [] }), + canReadAssurance + ? Promise.all([ + listAssuranceNodes(settings), + getAssuranceSummary(settings) + ]).then(([nodes, summary]) => ({ nodes: nodes.nodes, summary })) + : Promise.resolve({ nodes: [], summary: null }) ]); setSourcesAvailable(sources.available); setSourceSnapshots(sources.snapshots); setListSnapshots(lists.snapshots); setQueue(reviewQueue.candidates); + setAssuranceNodes(assurance.nodes); + setAssuranceSummary(assurance.summary); setSelectedCandidateId((current) => current && reviewQueue.candidates.some((item) => item.id === current) ? current : reviewQueue.candidates[0]?.id ?? "" ); + setSelectedAssuranceId((current) => + current || assurance.nodes[0]?.stable_id || "" + ); } catch (reason) { setError(errorMessage(reason)); } finally { setLoading(false); } - }, [canReview, settings]); + }, [canReadAssurance, canReview, settings]); useEffect(() => { void refresh(); @@ -125,6 +193,24 @@ export default function RiskCompliancePage({ }; }, [selectedCandidateId, settings]); + useEffect(() => { + if (!selectedAssuranceId || !canReadAssurance) { + setAssuranceGraph(null); + return; + } + let cancelled = false; + void getAssuranceGraph(settings, selectedAssuranceId) + .then((graph) => { + if (!cancelled) setAssuranceGraph(graph); + }) + .catch((reason) => { + if (!cancelled) setError(errorMessage(reason)); + }); + return () => { + cancelled = true; + }; + }, [canReadAssurance, selectedAssuranceId, settings]); + async function importSnapshot(item: ConnectorSnapshot) { setBusy(true); setError(""); @@ -182,6 +268,16 @@ export default function RiskCompliancePage({ ), disabled: !canReview + }, + { + id: "assurance", + label: ( + <> + + Assurance + + ), + disabled: !canReadAssurance } ]} /> @@ -250,6 +346,22 @@ export default function RiskCompliancePage({ settings={settings} /> )} + {view === "assurance" && ( + + )} ); @@ -754,6 +866,605 @@ function ReviewPane({ ); } +const ASSURANCE_STATES: Record = { + obligation: ["active", "suspended", "retired"], + governed_object: ["active", "inactive", "retired"], + risk: ["identified", "assessed", "accepted", "mitigated", "closed"], + control: [ + "designed", + "implemented", + "effective", + "failed", + "suspended", + "retired" + ], + evidence: ["current", "stale", "invalid", "superseded"], + finding: ["open", "accepted", "exception", "remediating", "resolved"], + corrective_measure: ["planned", "in_progress", "completed", "cancelled"], + effectiveness_review: ["pending", "effective", "ineffective", "inconclusive"] +}; + +const ASSURANCE_RELATIONS = [ + { id: "applies_to", source: "obligation", target: "governed_object" }, + { id: "exposes_risk", source: "governed_object", target: "risk" }, + { id: "mitigated_by", source: "risk", target: "control" }, + { id: "evidenced_by", source: "control", target: "evidence" }, + { id: "results_in", source: "evidence", target: "finding" }, + { id: "addressed_by", source: "finding", target: "corrective_measure" }, + { + id: "reviewed_by", + source: "corrective_measure", + target: "effectiveness_review" + } +] as const; + +type AssuranceNodeDraft = { + stableId: string; + kind: AssuranceNodeKind; + label: string; + description: string; + state: string; + ownerRef: string; + scopeRef: string; + governedObjectRef: string; + validFrom: string; + validTo: string; + classification: string; + legalBasisRefs: string; + policyRefs: string; + evidenceRefs: string; +}; + +function AssurancePane({ + nodes, + summary, + graph, + selectedId, + canWrite, + busy, + settings, + onSelect, + onBusy, + onError, + onNotice, + onRefresh +}: { + nodes: AssuranceNode[]; + summary: AssuranceSummary | null; + graph: { nodes: AssuranceNode[]; edges: AssuranceEdge[]; truncated: boolean } | null; + selectedId: string; + canWrite: boolean; + busy: boolean; + settings: PlatformRouteContext["settings"]; + onSelect: (id: string) => void; + onBusy: (value: boolean) => void; + onError: (message: string) => void; + onNotice: (message: string) => void; + onRefresh: () => Promise; +}) { + const [query, setQuery] = useState(""); + const [kind, setKind] = useState(""); + const [nodeDialogOpen, setNodeDialogOpen] = useState(false); + const [edgeDialogOpen, setEdgeDialogOpen] = useState(false); + const [editingNode, setEditingNode] = useState(null); + const [nodeDraft, setNodeDraft] = useState( + emptyAssuranceNodeDraft() + ); + const [edgeRelation, setEdgeRelation] = useState(""); + const [edgeTarget, setEdgeTarget] = useState(""); + const selected = nodes.find((item) => item.stable_id === selectedId) ?? null; + const visibleNodes = useMemo(() => { + const needle = query.trim().toLocaleLowerCase(); + return nodes.filter( + (item) => + (!kind || item.kind === kind) && + (!needle || + item.label.toLocaleLowerCase().includes(needle) || + item.stable_id.toLocaleLowerCase().includes(needle) || + (item.description || "").toLocaleLowerCase().includes(needle)) + ); + }, [kind, nodes, query]); + const availableRelations = selected + ? ASSURANCE_RELATIONS.filter((item) => item.source === selected.kind) + : []; + const activeRelation = availableRelations.find( + (item) => item.id === edgeRelation + ); + const edgeTargets = activeRelation + ? nodes.filter((item) => item.kind === activeRelation.target) + : []; + + function openNewNode() { + setEditingNode(null); + setNodeDraft(emptyAssuranceNodeDraft()); + setNodeDialogOpen(true); + } + + function openEditNode(item: AssuranceNode) { + setEditingNode(item); + setNodeDraft(nodeDraftFromItem(item)); + setNodeDialogOpen(true); + } + + async function submitNode(event: FormEvent) { + event.preventDefault(); + onBusy(true); + onError(""); + try { + const saved = await saveAssuranceNode( + settings, + assuranceNodeWrite(nodeDraft, editingNode?.provenance), + editingNode?.revision + ); + setNodeDialogOpen(false); + onSelect(saved.stable_id); + onNotice( + editingNode + ? `Recorded assurance revision ${saved.revision}.` + : "Created the assurance object." + ); + await onRefresh(); + } catch (reason) { + onError(errorMessage(reason)); + } finally { + onBusy(false); + } + } + + function openEdgeDialog() { + const first = availableRelations[0]; + setEdgeRelation(first?.id ?? ""); + setEdgeTarget( + first + ? nodes.find((item) => item.kind === first.target)?.stable_id ?? "" + : "" + ); + setEdgeDialogOpen(true); + } + + async function submitEdge(event: FormEvent) { + event.preventDefault(); + if (!selected || !activeRelation || !edgeTarget) return; + onBusy(true); + onError(""); + const value: AssuranceEdgeWrite = { + stable_id: `edge-${crypto.randomUUID()}`, + source_node_ref: selected.stable_id, + target_node_ref: edgeTarget, + relation: activeRelation.id, + state: "active", + owner_ref: selected.owner_ref, + scope_ref: selected.scope_ref, + valid_from: new Date().toISOString(), + valid_to: null, + provenance: { source: "risk-compliance-ui" }, + legal_basis_refs: selected.legal_basis_refs, + policy_refs: selected.policy_refs, + evidence_refs: [] + }; + try { + await saveAssuranceEdge(settings, value); + setEdgeDialogOpen(false); + onNotice("Connected the assurance objects."); + await onRefresh(); + } catch (reason) { + onError(errorMessage(reason)); + } finally { + onBusy(false); + } + } + + return ( +
+
+ {summary?.node_count ?? 0}objects + {summary?.edge_count ?? 0}relationships + {summary?.by_kind.risk ?? 0}risks + {summary?.by_state.open ?? 0}open findings +
+
+ +
+
+
+ {selected?.label || "Assurance object"} + {selected ? formatToken(selected.kind) : "Select an object"} +
+ {selected && canWrite && !selected.stable_id.startsWith("sanctions-") && ( + } + onClick={() => openEditNode(selected)} + /> + )} +
+ {!selected && ( +
Select an assurance object.
+ )} + {selected && ( +
+
+
State
+
Owner
{selected.owner_ref}
+
Scope
{selected.scope_ref || "Tenant"}
+
Valid from
{formatDate(selected.valid_from)}
+ {selected.governed_object_ref && ( +
Governed object
{selected.governed_object_ref}
+ )} +
Classification
{selected.classification}
+
+ {selected.description &&

{selected.description}

} +
+
+ Relationships + {graph?.edges.length ?? 0} in the bounded graph +
+ {canWrite && availableRelations.length > 0 && ( + + )} +
+
+ {graph?.edges.map((edge) => ( + + ))} + {!graph?.edges.length && ( +
No relationships are recorded.
+ )} +
+ {graph?.truncated && ( +
+ The bounded graph contains additional relationships. +
+ )} +
+ )} +
+
+ setNodeDialogOpen(false)} + onSubmit={submitNode} + /> + setEdgeDialogOpen(false)} + closeDisabled={busy} + footer={ + <> + + + + } + > +
+ + + + + + +
+ +
+ ); +} + +function AssuranceEdgeRow({ + edge, + nodes, + selectedId, + onSelect +}: { + edge: AssuranceEdge; + nodes: AssuranceNode[]; + selectedId: string; + onSelect: (id: string) => void; +}) { + const otherId = edge.source_node_ref === selectedId + ? edge.target_node_ref + : edge.source_node_ref; + const other = nodes.find((item) => item.stable_id === otherId); + return ( + + ); +} + +function AssuranceNodeDialog({ + open, + busy, + editing, + draft, + onChange, + onClose, + onSubmit +}: { + open: boolean; + busy: boolean; + editing: boolean; + draft: AssuranceNodeDraft; + onChange: (value: AssuranceNodeDraft) => void; + onClose: () => void; + onSubmit: (event: FormEvent) => void; +}) { + const update = (values: Partial) => + onChange({ ...draft, ...values }); + return ( + + + + + } + > +
+
+ + update({ stableId: event.target.value })} + disabled={editing} + required + maxLength={255} + /> + + + + + + update({ label: event.target.value })} required maxLength={500} /> + + + + + + update({ ownerRef: event.target.value })} required maxLength={500} /> + + + update({ scopeRef: event.target.value })} maxLength={500} /> + + {draft.kind === "governed_object" && ( + + update({ governedObjectRef: event.target.value })} required maxLength={1000} /> + + )} + + + + + update({ validFrom: event.target.value })} required /> + + + update({ validTo: event.target.value })} /> + +
+ +