From 50ae9e94d196f2b0db0d534d174eb36291095274 Mon Sep 17 00:00:00 2001 From: Benjamin Admin Date: Fri, 26 Jun 2026 10:58:00 +0200 Subject: [PATCH] feat(interpretation-in-map): judge a customer interpretation within the map (step 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Thin adapter — it judges the customer's reading WITHIN the already-built RegulatoryMap, it does not assess abstract legal questions and it is not RCI. - Reuses the existing assess_interpretation (no new legal reasoning); the 6 verdicts (plausible/too_narrow/too_broad/partially_correct/unsupported/uncertain) pass through unchanged. - Restricts affected_regulations/affected_obligations to those present in the map (intersection); links to the map's uncertain regulations. - Touched unsupported domains (wastewater/chemicals/...) are reported as future_corpus_domains (future_corpus_needed) — never pseudo-evaluated. - Customer-readable explanation ("Ihre Interpretation ist wahrscheinlich zu eng. … Betroffen in Ihrer Map: CRA."). - POST /reasoning/interpretation-in-map (renders the map, then interprets). - 7 tests; 63 green (existing reasoning MVP stays green), mypy clean, LOC ok. Co-Authored-By: Claude Opus 4.7 --- .../compliance/api/reasoning_routes.py | 12 ++ .../compliance/interpretation_map/__init__.py | 18 +++ .../compliance/interpretation_map/adapter.py | 90 +++++++++++ .../compliance/interpretation_map/schemas.py | 36 +++++ .../tests/test_interpretation_in_map.py | 141 ++++++++++++++++++ 5 files changed, 297 insertions(+) create mode 100644 backend-compliance/compliance/interpretation_map/__init__.py create mode 100644 backend-compliance/compliance/interpretation_map/adapter.py create mode 100644 backend-compliance/compliance/interpretation_map/schemas.py create mode 100644 backend-compliance/tests/test_interpretation_in_map.py diff --git a/backend-compliance/compliance/api/reasoning_routes.py b/backend-compliance/compliance/api/reasoning_routes.py index 0c734a66..f43538f2 100644 --- a/backend-compliance/compliance/api/reasoning_routes.py +++ b/backend-compliance/compliance/api/reasoning_routes.py @@ -9,12 +9,18 @@ pure deterministic rule evaluation. 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 + POST /reasoning/interpretation-in-map -> judge a customer interpretation within the map """ from __future__ import annotations from fastapi import APIRouter +from compliance.interpretation_map import ( + InterpretationInMapRequest, + InterpretationInMapResult, + interpret_in_map, +) from compliance.product_scope import ( ProductScopeRequest, ProductScopeResponse, @@ -71,6 +77,12 @@ def regulatory_map(req: RegulatoryMapRequest) -> RegulatoryMap: return render_regulatory_map(req.product_profile) +@router.post("/interpretation-in-map", response_model=InterpretationInMapResult) +def interpretation_in_map(req: InterpretationInMapRequest) -> InterpretationInMapResult: + reg_map = render_regulatory_map(req.product_profile) + return interpret_in_map(reg_map, req.customer_interpretation) + + @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/interpretation_map/__init__.py b/backend-compliance/compliance/interpretation_map/__init__.py new file mode 100644 index 00000000..0716d88d --- /dev/null +++ b/backend-compliance/compliance/interpretation_map/__init__.py @@ -0,0 +1,18 @@ +"""Interpretation-in-Map — evaluate a customer interpretation within the map. + +Thin adapter over the existing `assess_interpretation`: it judges the customer's +reading against the regulations/obligations actually present in the product's +RegulatoryMap, and flags touched unsupported domains as future_corpus_needed +instead of pseudo-evaluating them. No new legal reasoning, no RCI, no UI. +""" + +from __future__ import annotations + +from .adapter import interpret_in_map +from .schemas import InterpretationInMapRequest, InterpretationInMapResult + +__all__ = [ + "interpret_in_map", + "InterpretationInMapRequest", + "InterpretationInMapResult", +] diff --git a/backend-compliance/compliance/interpretation_map/adapter.py b/backend-compliance/compliance/interpretation_map/adapter.py new file mode 100644 index 00000000..e95cad32 --- /dev/null +++ b/backend-compliance/compliance/interpretation_map/adapter.py @@ -0,0 +1,90 @@ +"""Interpretation-in-Map adapter (step 5). + +Evaluates a customer interpretation WITHIN the already-built RegulatoryMap. It +reuses the existing `assess_interpretation` (no new legal engine), restricts the +affected regulations/obligations to those present in the map, and reports any +touched unsupported domain (wastewater/chemicals/...) as future_corpus_needed +rather than pseudo-evaluating it. +""" + +from __future__ import annotations + +from typing import Dict, List + +from compliance.reasoning.enums import InterpretationVerdict +from compliance.reasoning.interpretation_engine import assess_interpretation +from compliance.regulatory_map.schemas import RegulatoryMap + +from .schemas import InterpretationInMapResult + +_LABEL: Dict[InterpretationVerdict, str] = { + InterpretationVerdict.PLAUSIBLE: "plausibel", + InterpretationVerdict.TOO_NARROW: "zu eng", + InterpretationVerdict.TOO_BROAD: "zu weit", + InterpretationVerdict.PARTIALLY_CORRECT: "teilweise korrekt", + InterpretationVerdict.UNSUPPORTED: "nicht belegt", + InterpretationVerdict.UNCERTAIN: "unsicher", +} + +# domain -> keywords that signal the interpretation is ABOUT that (uncovered) domain. +_ENV_KEYWORDS: Dict[str, List[str]] = { + "environment_water": ["abwasser", "wastewater", "gewässer", "gewaesser", "einleitung", "abfluss"], + "chemicals": ["chemikalie", "reach", "clp", "reinigungsmittel", "biozid", "gefahrstoff", "detergenz", "lösemittel", "loesemittel"], + "environment_air": ["luft", "emission", "voc", "immission", "abluft", "verbrennung"], + "waste": ["abfall", "entsorgung", "weee", "recycling"], + "energy_resources": ["energie", "ökodesign", "oekodesign", "verbrauch"], +} + + +def _touches(text: str, domain: str) -> bool: + low = text.lower() + return any(kw in low for kw in _ENV_KEYWORDS.get(domain, [])) + + +def _explain(label: str, detail: str, affected_regs: List[str], future_domains: List[str], in_scope: bool) -> str: + base = "Ihre Interpretation ist wahrscheinlich %s." % label + if detail: + base += " " + detail + if affected_regs: + base += " Betroffen in Ihrer Map: %s." % ", ".join(affected_regs) + if future_domains: + base += ( + " Für %s liegt noch kein Regelkorpus vor — diese Aspekte werden nicht bewertet (future_corpus_needed)." + % ", ".join(future_domains) + ) + if not in_scope and not future_domains: + base += " Diese Auslegung betrifft kein Regelwerk Ihrer aktuellen Produkt-Map." + return base + + +def interpret_in_map(reg_map: RegulatoryMap, interpretation: str) -> InterpretationInMapResult: + a = assess_interpretation(interpretation) # existing engine — no new reasoning + + map_reg_ids = ( + {v.regulation_id for v in reg_map.applicable_regulations} + | {v.regulation_id for v in reg_map.uncertain_regulations} + | {v.regulation_id for v in reg_map.excluded_regulations} + ) + map_ob_ids = {o.obligation_id for v in reg_map.applicable_regulations for o in v.obligations} + uncertain_ids = {v.regulation_id for v in reg_map.uncertain_regulations} + + affected_regs = [r for r in a.affected_regulations if r in map_reg_ids] + affected_obs = [o for o in a.affected_obligations if o in map_ob_ids] + related_unc = [r for r in a.affected_regulations if r in uncertain_ids] + future = [d for d in reg_map.unsupported_domains if _touches(interpretation, d.domain)] + in_scope = bool(affected_regs or affected_obs) + + return InterpretationInMapResult( + raw_interpretation=interpretation, + assessment=a.assessment, + in_scope_of_map=in_scope, + affected_regulations=affected_regs, + affected_obligations=affected_obs, + related_uncertainties=related_unc, + future_corpus_domains=future, + corrected_interpretation=a.corrected_interpretation, + risks=a.risks, + legal_basis_refs=a.legal_basis_refs, + explanation=_explain(_LABEL[a.assessment], a.explanation, affected_regs, [d.domain for d in future], in_scope), + confidence=a.confidence, + ) diff --git a/backend-compliance/compliance/interpretation_map/schemas.py b/backend-compliance/compliance/interpretation_map/schemas.py new file mode 100644 index 00000000..99362bd0 --- /dev/null +++ b/backend-compliance/compliance/interpretation_map/schemas.py @@ -0,0 +1,36 @@ +"""Schemas for Interpretation-in-Map (step 5). + +A thin adapter that evaluates a customer interpretation WITHIN the already-built +RegulatoryMap — it does not assess abstract legal questions. Application types +only; no compliance-meta-model classes (freeze v1.0 untouched). +""" + +from __future__ import annotations + +from typing import List + +from pydantic import BaseModel, Field + +from compliance.product_scope.schemas import UnsupportedDomain +from compliance.profile.canonical import CanonicalProductRegulatoryProfile +from compliance.reasoning.enums import Confidence, InterpretationVerdict + + +class InterpretationInMapRequest(BaseModel): + product_profile: CanonicalProductRegulatoryProfile + customer_interpretation: str + + +class InterpretationInMapResult(BaseModel): + raw_interpretation: str + assessment: InterpretationVerdict + in_scope_of_map: bool # True if it touches a regulation/obligation present in the map + affected_regulations: List[str] = Field(default_factory=list) # intersected with the map + affected_obligations: List[str] = Field(default_factory=list) # intersected (registry-linked) + related_uncertainties: List[str] = Field(default_factory=list) # map-uncertain regs it touches + future_corpus_domains: List[UnsupportedDomain] = Field(default_factory=list) # NOT evaluated + corrected_interpretation: str = "" + risks: List[str] = Field(default_factory=list) + legal_basis_refs: List[str] = Field(default_factory=list) + explanation: str = "" + confidence: Confidence = Confidence.MEDIUM diff --git a/backend-compliance/tests/test_interpretation_in_map.py b/backend-compliance/tests/test_interpretation_in_map.py new file mode 100644 index 00000000..0e6df9bb --- /dev/null +++ b/backend-compliance/tests/test_interpretation_in_map.py @@ -0,0 +1,141 @@ +"""Tests for Interpretation-in-Map (step 5). + +Acceptance: a customer interpretation is judged against the existing map, using +only assess_interpretation; affected regulations/obligations are referenced from +the map; unsupported domains (wastewater/chemicals) are flagged +future_corpus_needed, not pseudo-evaluated; output is customer-readable. +""" + +from __future__ import annotations + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from compliance.interpretation_map import interpret_in_map +from compliance.profile.canonical import ( + CanonicalLifecyclePhase, + CanonicalProductRegulatoryProfile, + CanonicalProductType, + EconomicOperatorRole, + EnvironmentalImpact, +) +from compliance.reasoning.enums import InterpretationVerdict +from compliance.reasoning.interpretation_engine import assess_interpretation +from compliance.regulatory_map import render_regulatory_map + + +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) + + +def _map(**ov): + return render_regulatory_map(ready_profile(**ov)) + + +# 1 + 2. evaluated against the map, using ONLY assess_interpretation. +def test_uses_assess_interpretation_verdict(): + text = "Wir glauben, der CRA gilt nur für neue Produkte." + result = interpret_in_map(_map(), text) + assert result.assessment == assess_interpretation(text).assessment == InterpretationVerdict.TOO_NARROW + assert "CRA" in result.affected_regulations # CRA is in the map + assert result.in_scope_of_map is True + + +# 3. the six verdict values pass through unchanged. +def test_verdict_values(): + m = _map() + assert interpret_in_map(m, "CRA gilt nur für neue Produkte.").assessment == InterpretationVerdict.TOO_NARROW + assert interpret_in_map(m, "Open Source ist ausgenommen, also betrifft uns der CRA nicht.").assessment == InterpretationVerdict.PARTIALLY_CORRECT + assert interpret_in_map(m, "Der Mond beeinflusst unsere Updatezyklen.").assessment == InterpretationVerdict.UNCERTAIN + + +# 4. affected regulations/obligations are referenced FROM the map. +def test_affected_refs_from_map(): + m = _map() + result = interpret_in_map(m, "Eine SBOM reicht, dann sind wir fertig.") + map_ob_ids = {o.obligation_id for v in m.applicable_regulations for o in v.obligations} + map_reg_ids = {v.regulation_id for v in m.applicable_regulations} | {v.regulation_id for v in m.uncertain_regulations} + assert "sbom_creation" in result.affected_obligations + assert set(result.affected_obligations) <= map_ob_ids + assert set(result.affected_regulations) <= map_reg_ids + + +# 5. environmental aspects are NOT pseudo-evaluated. +def test_environmental_not_pseudo_evaluated(): + m = _map(environmental=EnvironmentalImpact(discharges_to_wastewater=True)) + result = interpret_in_map(m, "Beim Abwasser sind wir nicht betroffen, das spielt für uns keine Rolle.") + domains = {d.domain for d in result.future_corpus_domains} + assert "environment_water" in domains + assert "future_corpus_needed" in result.explanation + + +# 6. output is customer-readable. +def test_customer_readable(): + result = interpret_in_map(_map(), "Der CRA gilt nur für neue Produkte.") + assert "zu eng" in result.explanation + assert result.explanation.startswith("Ihre Interpretation ist wahrscheinlich") + + +# affected refs never leave the map (no abstract legal questions). +def test_affected_regs_never_outside_map(): + m = _map() + map_reg_ids = ( + {v.regulation_id for v in m.applicable_regulations} + | {v.regulation_id for v in m.uncertain_regulations} + | {v.regulation_id for v in m.excluded_regulations} + ) + for text in ["CRA gilt nur für neue Produkte.", "Ohne Funkmodul keine Cyber-Pflichten.", "SBOM reicht."]: + result = interpret_in_map(m, text) + assert set(result.affected_regulations) <= map_reg_ids + + +# 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_interpretation_in_map(client): + r = client.post( + "/reasoning/interpretation-in-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"], + }, + "customer_interpretation": "Der CRA gilt nur für neue Produkte.", + }, + ) + assert r.status_code == 200 + body = r.json() + assert body["assessment"] == "too_narrow" + assert "CRA" in body["affected_regulations"] + assert "zu eng" in body["explanation"]