feat(reasoning): Regulatory Reasoning Engine MVP (scope/obligations/implementation/interpretation)
Deterministic reasoning layer ON TOP of the Legal Knowledge Graph (obligation
registry) and the Compliance Execution Graph (control mapping/evidence). Answers
which regulations apply to a concrete product, which obligations follow, whether
the customer's implementation covers them, and whether a customer interpretation
is too narrow/broad/plausible.
- ProductProfile with tri-state facts (Optional[bool]=None => uncertain, never
false security); safe predicate evaluator (no eval).
- 6 regulation triggers (CRA/MaschinenVO/RED/EMV/DataAct/NIS2) with missing-fact
prompts; 24 obligation scope rules.
- CRA obligation_ids RE-USED verbatim from the registry (93 ids) — never re-minted
(control_uuid trap); Machine/Data-Act flagged proposed=True.
- required_evidence constrained to the framework-agnostic shared evidence catalog;
capabilities echo the planned Obligation->Capability layer.
- Overlap groups (CRA<->MaschinenVO cyber-safety) + evidence-for-multiple (USP).
- 4 endpoints POST /reasoning/{scope,obligations,implementation-assessment,
interpretation-assessment}; thin handlers, registered in api/__init__.py.
- 22 tests (5 machine-builder scenarios + 10 acceptance questions). No DB
migration, no RAG, no new controls.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,100 @@
|
||||
"""Safe, tri-state condition evaluator for applicability rules.
|
||||
|
||||
Conditions are plain data (no `eval`): a *leaf* is a 3-tuple
|
||||
``(field, op, value)``; a *composite* is ``{"all": [...]}`` or
|
||||
``{"any": [...]}``. Evaluation is tri-state — ``True`` / ``False`` /
|
||||
``None`` (unknown) — so a missing product fact yields *uncertain*, never a
|
||||
false negative.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import Enum
|
||||
from typing import Any, Dict, List, Optional, Tuple, Union
|
||||
|
||||
Leaf = Tuple[str, str, Any]
|
||||
Condition = Union[Leaf, Dict[str, Any]]
|
||||
|
||||
|
||||
def _attr(profile: Any, field: str) -> Any:
|
||||
value = getattr(profile, field, None)
|
||||
if isinstance(value, Enum):
|
||||
return value.value
|
||||
return value
|
||||
|
||||
|
||||
def _eval_leaf(leaf: Leaf, profile: Any) -> Optional[bool]:
|
||||
field, op, expected = leaf
|
||||
actual = _attr(profile, field)
|
||||
|
||||
if op == "not_none":
|
||||
return actual is not None
|
||||
if op == "is_none":
|
||||
return actual is None
|
||||
|
||||
if op == "contains_any":
|
||||
# list-valued field (e.g. product_type); empty list = known-empty.
|
||||
items = actual or []
|
||||
hay = " ".join(str(x).lower() for x in items)
|
||||
return any(str(k).lower() in hay for k in expected)
|
||||
|
||||
if actual is None:
|
||||
return None # unknown fact -> unknown result
|
||||
|
||||
if op == "eq":
|
||||
return bool(actual == expected)
|
||||
if op == "ne":
|
||||
return bool(actual != expected)
|
||||
if op == "truthy":
|
||||
return bool(actual)
|
||||
if op == "falsy":
|
||||
return not bool(actual)
|
||||
if op == "in":
|
||||
return bool(actual in expected)
|
||||
if op == "not_in":
|
||||
return bool(actual not in expected)
|
||||
if op == "date_after":
|
||||
return bool(actual > expected)
|
||||
raise ValueError("unknown predicate op: %r" % (op,))
|
||||
|
||||
|
||||
def evaluate(condition: Optional[Condition], profile: Any) -> Optional[bool]:
|
||||
"""Return True/False/None(unknown) for a condition tree."""
|
||||
if condition is None:
|
||||
return True
|
||||
if isinstance(condition, tuple):
|
||||
return _eval_leaf(condition, profile)
|
||||
|
||||
if "all" in condition:
|
||||
results = [evaluate(c, profile) for c in condition["all"]]
|
||||
if any(r is False for r in results):
|
||||
return False
|
||||
if any(r is None for r in results):
|
||||
return None
|
||||
return True
|
||||
if "any" in condition:
|
||||
results = [evaluate(c, profile) for c in condition["any"]]
|
||||
if any(r is True for r in results):
|
||||
return True
|
||||
if any(r is None for r in results):
|
||||
return None
|
||||
return False
|
||||
raise ValueError("malformed condition: %r" % (condition,))
|
||||
|
||||
|
||||
def true_leaves(condition: Optional[Condition], profile: Any) -> List[Leaf]:
|
||||
"""Collect the leaf conditions that evaluated True (for trigger_facts)."""
|
||||
if condition is None:
|
||||
return []
|
||||
if isinstance(condition, tuple):
|
||||
return [condition] if _eval_leaf(condition, profile) is True else []
|
||||
members = condition.get("all") or condition.get("any") or []
|
||||
out: List[Leaf] = []
|
||||
for c in members:
|
||||
out.extend(true_leaves(c, profile))
|
||||
return out
|
||||
|
||||
|
||||
def unknown_fields(fields: List[str], profile: Any) -> List[str]:
|
||||
"""Subset of `fields` whose value on the profile is None (unknown)."""
|
||||
return [f for f in fields if _attr(profile, f) is None]
|
||||
Reference in New Issue
Block a user