diff --git a/backend-compliance/compliance/api/reasoning_routes.py b/backend-compliance/compliance/api/reasoning_routes.py index 49e3f305..0c734a66 100644 --- a/backend-compliance/compliance/api/reasoning_routes.py +++ b/backend-compliance/compliance/api/reasoning_routes.py @@ -8,6 +8,7 @@ pure deterministic rule evaluation. POST /reasoning/implementation-reasoning -> claim->obligation mapping (Welt 1, no verdict) POST /reasoning/interpretation-assessment -> verdict on a customer interpretation POST /reasoning/product-scope -> gate on facts, else run discover_scope once + POST /reasoning/regulatory-map -> customer-readable read-model over the scope """ from __future__ import annotations @@ -19,6 +20,7 @@ from compliance.product_scope import ( ProductScopeResponse, resolve_product_scope, ) +from compliance.regulatory_map import RegulatoryMap, RegulatoryMapRequest, render_regulatory_map from compliance.reasoning import ( assess_interpretation, derive_obligations, @@ -64,6 +66,11 @@ def product_scope(req: ProductScopeRequest) -> ProductScopeResponse: return resolve_product_scope(req.product_profile) +@router.post("/regulatory-map", response_model=RegulatoryMap) +def regulatory_map(req: RegulatoryMapRequest) -> RegulatoryMap: + return render_regulatory_map(req.product_profile) + + @router.post("/interpretation-assessment", response_model=InterpretationResponse) def interpretation_assessment(req: InterpretationRequest) -> InterpretationResponse: result = assess_interpretation(req.customer_interpretation, req.product_profile) diff --git a/backend-compliance/compliance/regulatory_map/__init__.py b/backend-compliance/compliance/regulatory_map/__init__.py new file mode 100644 index 00000000..50bb97ee --- /dev/null +++ b/backend-compliance/compliance/regulatory_map/__init__.py @@ -0,0 +1,31 @@ +"""Regulatory Map — customer-readable read-model over the engine's scope output. + +Composes scope + registry-linked obligations + overlaps into one map: +product -> trigger facts -> applicable / uncertain / excluded regulations -> +obligations -> overlaps -> unsupported domains -> executive summary. Explains the +engine's state, never extends it. No new logic, no UI, no RAG, no percentage. +""" + +from __future__ import annotations + +from .renderer import render_regulatory_map +from .schemas import ( + ApplicableRegulationView, + ExcludedRegulationView, + ObligationRef, + OverlapView, + RegulatoryMap, + RegulatoryMapRequest, + UncertainRegulationView, +) + +__all__ = [ + "render_regulatory_map", + "RegulatoryMap", + "RegulatoryMapRequest", + "ApplicableRegulationView", + "UncertainRegulationView", + "ExcludedRegulationView", + "OverlapView", + "ObligationRef", +] diff --git a/backend-compliance/compliance/regulatory_map/renderer.py b/backend-compliance/compliance/regulatory_map/renderer.py new file mode 100644 index 00000000..af0cf484 --- /dev/null +++ b/backend-compliance/compliance/regulatory_map/renderer.py @@ -0,0 +1,169 @@ +"""Regulatory Map renderer (step 4) — pure composition, no new logic. + +It explains the engine's state, it does not extend it: every statement comes +from `resolve_product_scope` (scope verdict) or `derive_obligations` (registry- +linked obligations + overlaps). No legal decisions here; obligations are shown +ONLY where a registry id is linkable (registry_anchor); the executive summary +carries counts but NO percentage. +""" + +from __future__ import annotations + +from typing import Dict, List + +from compliance.navigator.engine import navigate +from compliance.product_scope.orchestrator import resolve_product_scope +from compliance.product_scope.schemas import RegulatoryScopeResult, ScopeStatus +from compliance.profile.canonical import CanonicalProductRegulatoryProfile +from compliance.profile.to_reasoning import to_reasoning_profile +from compliance.reasoning.obligation_engine import derive_obligations + +from .schemas import ( + ApplicableRegulationView, + ExcludedRegulationView, + ObligationRef, + OverlapView, + RegulatoryMap, + UncertainRegulationView, +) + +_DOMAIN_BY_REG = { + "CRA": "cyber", + "MaschinenVO": "machine_safety", + "RED": "radio", + "DataAct": "data", + "EMV": "emv", + "NIS2": None, +} + + +def _product_summary(c: CanonicalProductRegulatoryProfile) -> str: + bits: List[str] = [c.name or "Produkt"] + if c.product_type: + bits.append("(%s)" % c.product_type.value) + sig: List[str] = [] + if c.is_machine: + sig.append("Maschine") + if c.has_remote_access or c.connected_to_internet or "cloud" in c.technologies: + sig.append("vernetzt") + if c.has_embedded_software: + sig.append("Firmware") + if c.economic_operator_role: + sig.append("Rolle: %s" % c.economic_operator_role.value) + if c.markets: + sig.append("Märkte: %s" % ", ".join(c.markets)) + if sig: + bits.append("— " + "; ".join(sig)) + return " ".join(bits) + + +def render_regulatory_map(profile: CanonicalProductRegulatoryProfile) -> RegulatoryMap: + scope_resp = resolve_product_scope(profile) + summary = _product_summary(profile) + + if scope_resp.status == ScopeStatus.NEEDS_FACTS: + return RegulatoryMap( + scope_resolved=False, + product_summary=summary, + executive_summary=( + "Regulatorischer Scope noch nicht bestimmbar — zuerst Mindestfakten klären: " + + "; ".join(scope_resp.missing_facts[:6]) + + "." + ), + ) + + scope = scope_resp.regulatory_scope + assert scope is not None + obligations = derive_obligations(to_reasoning_profile(profile)) + nav_questions = navigate(profile).suggested_questions + + linked_ids = {o.obligation_id for o in obligations.applicable_obligations if o.registry_anchor} + by_reg: Dict[str, List[ObligationRef]] = {} + shared_ev: Dict[str, List[str]] = {} + for o in obligations.applicable_obligations: + if not o.registry_anchor: + continue + by_reg.setdefault(o.source_regulation, []).append( + ObligationRef( + obligation_id=o.obligation_id, + title=o.title, + legal_basis_refs=o.legal_basis_refs, + authority_level=o.authority_level, + ) + ) + for ev in o.required_evidence: + shared_ev.setdefault(ev, []).append(o.obligation_id) + + applicable_views = [] + for r in scope.applicable_regulations: + obs = by_reg.get(r.regulation_id, []) + applicable_views.append( + ApplicableRegulationView( + regulation_id=r.regulation_id, + name=r.name, + why_applicable=r.explanation, + triggered_by=r.trigger_facts, + obligations=obs, + obligations_note="" if obs else "Pflichten für dieses Regelwerk sind noch nicht registry-verlinkt.", + confidence=r.confidence, + ) + ) + + uncertain_views = [] + for u in scope.uncertain_regulations: + domain = _DOMAIN_BY_REG.get(u.regulation_id) + qrefs = [q.question_id for q in nav_questions if domain and domain in q.regulatory_domains_unblocked] + uncertain_views.append( + UncertainRegulationView( + regulation_id=u.regulation_id, name=u.name, missing_facts=u.missing_facts, question_refs=qrefs + ) + ) + + overlap_views = [] + for ov in obligations.overlaps: + members = [m for m in ov.obligations if m in linked_ids] + if len(members) >= 2: + overlap_views.append( + OverlapView(overlap_group_id=ov.overlap_group_id, shared_obligations=members, explanation=ov.explanation) + ) + + trigger_facts: List[str] = [] + for v in applicable_views: + for t in v.triggered_by: + if t not in trigger_facts: + trigger_facts.append(t) + + return RegulatoryMap( + scope_resolved=True, + product_summary=summary, + trigger_facts=trigger_facts, + applicable_regulations=applicable_views, + uncertain_regulations=uncertain_views, + excluded_regulations=[ + ExcludedRegulationView(regulation_id=e.regulation_id, name=e.name, exclusion_reason=e.reason) + for e in scope.excluded_regulations + ], + unsupported_domains=scope.unsupported_domains, + overlaps=overlap_views, + shared_evidence={ev: ids for ev, ids in shared_ev.items() if len(ids) > 1}, + executive_summary=_executive_summary(summary, applicable_views, uncertain_views, scope, len(linked_ids)), + ) + + +def _executive_summary( + summary: str, + applicable: List[ApplicableRegulationView], + uncertain: List[UncertainRegulationView], + scope: RegulatoryScopeResult, + n_obligations: int, +) -> str: + appl = ", ".join(v.regulation_id for v in applicable) or "—" + unc = ", ".join(v.regulation_id for v in uncertain) or "keine" + exc = ", ".join(e.regulation_id for e in scope.excluded_regulations) or "keine" + uns = ", ".join(d.domain for d in scope.unsupported_domains) or "keine" + return ( + "Für %s gelten nach derzeitigem Stand wahrscheinlich: %s. Unsicher (fehlende Fakten): %s. " + "Ausgeschlossen: %s. Nicht abgedeckt (Regelkorpus fehlt): %s. Ermittelt: %d registry-verlinkte " + "Pflichten. Es wurden keine weiteren Regelwerke im aktuellen Korpus identifiziert." + % (summary, appl, unc, exc, uns, n_obligations) + ) diff --git a/backend-compliance/compliance/regulatory_map/schemas.py b/backend-compliance/compliance/regulatory_map/schemas.py new file mode 100644 index 00000000..1b83661b --- /dev/null +++ b/backend-compliance/compliance/regulatory_map/schemas.py @@ -0,0 +1,70 @@ +"""Read-model for the Regulatory Map (step 4). + +A customer-readable view that COMPOSES what the engine already computed (scope + +obligations + overlaps). It adds no scope/obligation logic. All fields are +application-level presentation types — NOT compliance-meta-model classes +(architecture freeze v1.0 untouched). +""" + +from __future__ import annotations + +from typing import Dict, List + +from pydantic import BaseModel, Field + +from compliance.product_scope.schemas import UnsupportedDomain +from compliance.profile.canonical import CanonicalProductRegulatoryProfile +from compliance.reasoning.enums import AuthorityLevel, Confidence + + +class RegulatoryMapRequest(BaseModel): + product_profile: CanonicalProductRegulatoryProfile + + +class ObligationRef(BaseModel): + obligation_id: str + title: str + legal_basis_refs: List[str] = Field(default_factory=list) + authority_level: AuthorityLevel + + +class ApplicableRegulationView(BaseModel): + regulation_id: str + name: str + why_applicable: str + triggered_by: List[str] = Field(default_factory=list) + obligations: List[ObligationRef] = Field(default_factory=list) + obligations_note: str = "" # set when obligations are not yet registry-linkable + confidence: Confidence + + +class UncertainRegulationView(BaseModel): + regulation_id: str + name: str + missing_facts: List[str] = Field(default_factory=list) + question_refs: List[str] = Field(default_factory=list) + + +class ExcludedRegulationView(BaseModel): + regulation_id: str + name: str + exclusion_reason: str + + +class OverlapView(BaseModel): + overlap_group_id: str + shared_obligations: List[str] = Field(default_factory=list) + explanation: str = "" + + +class RegulatoryMap(BaseModel): + scope_resolved: bool + product_summary: str + trigger_facts: List[str] = Field(default_factory=list) + applicable_regulations: List[ApplicableRegulationView] = Field(default_factory=list) + uncertain_regulations: List[UncertainRegulationView] = Field(default_factory=list) + excluded_regulations: List[ExcludedRegulationView] = Field(default_factory=list) + unsupported_domains: List[UnsupportedDomain] = Field(default_factory=list) + overlaps: List[OverlapView] = Field(default_factory=list) + shared_evidence: Dict[str, List[str]] = Field(default_factory=dict) + executive_summary: str = "" diff --git a/backend-compliance/tests/test_regulatory_map.py b/backend-compliance/tests/test_regulatory_map.py new file mode 100644 index 00000000..75de2c7e --- /dev/null +++ b/backend-compliance/tests/test_regulatory_map.py @@ -0,0 +1,159 @@ +"""Tests for the Regulatory Map renderer (step 4). + +Acceptance: the renderer makes no own legal decisions (it composes the scope + +registry-linked obligations); CRA/MaschVO/EMV are separate; RED/DataAct/NIS2 are +uncertain; environmental is unsupported (not applicable); obligations appear only +when registry-linkable; the executive summary has no percentage. +""" + +from __future__ import annotations + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from compliance.product_scope import resolve_product_scope +from compliance.profile.canonical import ( + CanonicalLifecyclePhase, + CanonicalProductRegulatoryProfile, + CanonicalProductType, + EconomicOperatorRole, + EnvironmentalImpact, +) +from compliance.regulatory_map import render_regulatory_map + +_PROPOSED_IDS = { + "machine_risk_assessment", "machine_safety_control_systems", "machine_protection_against_corruption", + "machine_instructions_for_use", "machine_ce_conformity", "data_act_data_access_by_design", + "data_act_user_data_access", "cra_secure_by_design", "cra_risk_assessment", + "cra_technical_documentation", "cra_ce_conformity_assessment", "cra_instructions_for_use", +} + + +def ready_profile(**ov) -> CanonicalProductRegulatoryProfile: + base = dict( + name="Industriespülmaschine", + product_type=CanonicalProductType.MACHINERY, + markets=["EU", "DE"], + economic_operator_role=EconomicOperatorRole.MANUFACTURER, + lifecycle_phase=CanonicalLifecyclePhase.PLACING_ON_MARKET, + is_machine=True, + is_component=False, + has_software_updates=True, + has_embedded_software=True, + has_remote_access=True, + technologies=["cloud", "ota_updates"], + ) + base.update(ov) + return CanonicalProductRegulatoryProfile(**base) + + +# 1. renderer makes no own decisions — it mirrors the scope verdict exactly. +def test_no_own_legal_decisions(): + p = ready_profile() + m = render_regulatory_map(p) + scope = resolve_product_scope(p).regulatory_scope + assert {v.regulation_id for v in m.applicable_regulations} == { + r.regulation_id for r in scope.applicable_regulations + } + assert {v.regulation_id for v in m.uncertain_regulations} == { + r.regulation_id for r in scope.uncertain_regulations + } + + +# 2/3/5. CRA/MaschVO/EMV separate applicable; RED/DataAct/NIS2 uncertain. +def test_regulation_separation(): + m = render_regulatory_map(ready_profile()) + applicable = {v.regulation_id for v in m.applicable_regulations} + uncertain = {v.regulation_id for v in m.uncertain_regulations} + assert {"CRA", "MaschinenVO", "EMV"} <= applicable + assert {"RED", "DataAct", "NIS2"} <= uncertain + + +# 4. environmental triggers surface as unsupported_domain, never applicable. +def test_environmental_unsupported_not_applicable(): + p = ready_profile(environmental=EnvironmentalImpact(discharges_to_wastewater=True, uses_cleaning_chemicals=True)) + m = render_regulatory_map(p) + domains = {d.domain for d in m.unsupported_domains} + assert "environment_water" in domains and "chemicals" in domains + assert all(v.regulation_id in {"CRA", "MaschinenVO", "RED", "DataAct", "EMV", "NIS2"} for v in m.applicable_regulations) + + +# 6. obligations are shown only when a registry id is linkable. +def test_obligations_only_registry_linkable(): + m = render_regulatory_map(ready_profile()) + shown = {o.obligation_id for v in m.applicable_regulations for o in v.obligations} + assert shown # CRA registry obligations are shown + assert "sbom_creation" in shown + assert not (shown & _PROPOSED_IDS) # no proposed (non-registry) obligation leaks in + # MaschinenVO is applicable but its obligations are proposed -> empty + note + machvo = next(v for v in m.applicable_regulations if v.regulation_id == "MaschinenVO") + assert machvo.obligations == [] + assert machvo.obligations_note + + +# 7. executive summary contains no percentage. +def test_executive_summary_no_percent(): + m = render_regulatory_map(ready_profile()) + assert "%" not in m.executive_summary + assert "prozent" not in m.executive_summary.lower() + + +# 8. output is customer-readable and structured. +def test_customer_readable(): + m = render_regulatory_map(ready_profile()) + assert m.product_summary + assert "wahrscheinlich" in m.executive_summary + assert "Unsicher" in m.executive_summary + assert m.trigger_facts + + +# needs-facts profile -> map says scope not yet resolved. +def test_needs_facts_map(): + m = render_regulatory_map(CanonicalProductRegulatoryProfile(name="X")) + assert m.scope_resolved is False + assert "Mindestfakten" in m.executive_summary + assert m.applicable_regulations == [] + + +# uncertain RED links to the radio navigator question. +def test_uncertain_links_to_navigator_question(): + m = render_regulatory_map(ready_profile()) + red = next(v for v in m.uncertain_regulations if v.regulation_id == "RED") + assert "has_radio_module" in red.question_refs + + +# endpoint smoke. +@pytest.fixture(scope="module") +def client(): + from compliance.api.reasoning_routes import router + + app = FastAPI() + app.include_router(router) + return TestClient(app) + + +def test_endpoint_regulatory_map(client): + r = client.post( + "/reasoning/regulatory-map", + json={ + "product_profile": { + "name": "M", + "product_type": "machinery", + "markets": ["EU"], + "economic_operator_role": "manufacturer", + "lifecycle_phase": "placing_on_market", + "is_machine": True, + "is_component": False, + "has_software_updates": True, + "has_embedded_software": True, + "has_remote_access": True, + "technologies": ["cloud"], + } + }, + ) + assert r.status_code == 200 + body = r.json() + assert body["scope_resolved"] is True + assert {v["regulation_id"] for v in body["applicable_regulations"]} >= {"CRA", "MaschinenVO"} + assert "%" not in body["executive_summary"]