feat(interpretation-in-map): judge a customer interpretation within the map (step 5)

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 <noreply@anthropic.com>
2026-06-26 10:58:00 +02:00
parent 9312ad18ef
commit 50ae9e94d1
5 changed files with 297 additions and 0 deletions
@@ -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)
@@ -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",
+]
@@ -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,
+    )
@@ -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
@@ -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"]