feat(regulatory-map): customer-readable read-model over the scope (step 4)

The Map Renderer explains the engine's state, it does not extend it. Pure
composition of resolve_product_scope (scope verdict) + derive_obligations
(registry-linked obligations + overlaps) into one RegulatoryMap.

- product_summary, trigger_facts, applicable/uncertain/excluded regulations,
  unsupported_domains, overlaps (shared_obligations), shared_evidence, and a
  customer-readable executive_summary.
- No own legal decisions: applicable/uncertain mirror the scope verdict exactly.
- Obligations shown ONLY when registry-linkable (registry_anchor) — MaschinenVO/
  EMV obligations are proposed, so they render empty + a note, never as linked.
  Overlaps/shared_evidence likewise filtered to registry-linked members.
- Uncertain regulations link to the navigator question that would resolve them
  (RED -> has_radio_module, DataAct -> generates_usage_data).
- Environmental appears only as unsupported_domain; executive_summary has NO
  percentage (counts + "no further regulations identified" instead).
- POST /reasoning/regulatory-map (thin handler). Response types are presentation-
  level, not meta-model classes (freeze v1.0 untouched).
- 9 tests; 56 green (existing reasoning MVP stays green), mypy clean, LOC ok.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-06-26 10:36:06 +02:00
parent 4e8eb2dc0e
commit 9312ad18ef
5 changed files with 436 additions and 0 deletions
@@ -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)
@@ -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",
]
@@ -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)
)
@@ -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 = ""
@@ -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"]