""" MACI — Reconciliation Layer (Phase 5) -------------------------------------- Closes two gaps identified in pre-SRB-meeting adversarial testing (2026-08-10), against the live production API: GAP 1 — Recall: the base XLM-R classifier missed two real structural violations (Bay' al-Inah buy-back, risk/ownership decoupling) when presented in Arabic/Urdu, despite catching the English-phrased equivalents. The structural_patterns.py layer (Phase 4) already detects these correctly in all three languages — it just wasn't being used to CHANGE the final verdict, only to add supplementary flags alongside whatever the ML model said. GAP 2 — Precision: the base classifier false-positives on standard AAOIFI-compliant actual-loss indemnity language (Murabaha default compensation calculated as cost-price-vs-resale-price difference), flagging it as Gharar/HIGH. This is the exact clause shape AAOIFI Shari'a Standard No. 8 §7 explicitly permits. WHAT THIS MODULE DOES: Runs structural_patterns.scan_structural_patterns() and a new, narrow set of COMPLIANT hard-negative patterns alongside the base classifier, then reconciles the two signals into one verdict: 1. If a structural VIOLATION pattern fires (SP-01..SP-05), it wins, regardless of what the base classifier said. These are deterministic, SRB-derived patterns — higher trust than a probabilistic ML flag missing on this language. 2. If the base classifier flags a violation, no structural violation fired, AND a known compliant hard-negative pattern matches, the verdict is DOWNGRADED — not silently cleared. It becomes "REVIEW" with an explicit reconciliation note, never "AUTHENTIC" outright. A human still looks at it; the difference is that "definitely quarantine this" becomes "a human should confirm this, here is exactly why it might be a false positive." 3. Otherwise, the base classifier's verdict passes through unchanged. WHAT THIS MODULE DELIBERATELY DOES NOT DO: - It does not retrain or touch classifier weights. The false positive on clause type #7 is still present in the base model — this suppresses its effect on the final verdict for a known, narrow pattern, it does not fix the model's understanding. - It does not downgrade CRITICAL-severity ML flags. A pattern this narrow is not trusted to override the base model's highest- confidence violation category (Riba) even if surface phrasing happens to overlap — only HIGH/MEDIUM Gharar-type flags, which is where the observed false positive actually lives. - The hard-negative list starts at ONE pattern (the exact one found in testing), not a broad category. Expand it only against new, specifically observed false positives — a wide net here just trades a false-positive problem for a false-negative one. HONESTY NOTE — same posture as structural_patterns.py: This is a targeted patch for two specific, observed failure modes, not a general precision/recall fix. It should be presented as exactly that. """ import re from dataclasses import dataclass, field from typing import List, Optional from structural_patterns import ( scan_structural_patterns, normalize_arabic_script, StructuralFlag, ) @dataclass class HardNegativePattern: pattern_id: str name: str aaoifi_standard: str explanation: str all_of: List[str] = field(default_factory=list) # If any ML-flagged violation_class is NOT in this set, the # hard-negative is ignored — e.g. this pattern should only ever # suppress a Gharar flag, never a Riba flag, even if it matched. applies_to_violation_classes: List[str] = field(default_factory=list) # Never downgrade at this severity or above. max_ml_severity: str = "HIGH" # Severity ordering used for the max_ml_severity guard. _SEVERITY_RANK = {"NONE": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4} COMPLIANT_HARD_NEGATIVES: List[HardNegativePattern] = [ HardNegativePattern( pattern_id="CHN-01", name="Actual-Loss Indemnity (Cost-vs-Resale-Price Difference)", aaoifi_standard=( "AAOIFI Shari'a Standard No. 8 (Murabaha), §7 — compensation " "for actual loss on a defaulted purchase undertaking, " "calculated as a concrete formula rather than an open-ended " "or discretionary penalty, is permitted and is not Gharar." ), explanation=( "Clause compensates the seller for the difference between the " "cost price and the actual resale price obtained from a third " "party after a failed purchase undertaking. This is a fixed, " "calculable, after-the-fact formula tied to a real transaction " "— not an open-ended or discretionary amount — which is the " "specific feature that distinguishes permitted actual-loss " "indemnity from prohibited Gharar-bearing penalty clauses. " "This is standard, AAOIFI-compliant Murabaha default-" "compensation language, not excessive uncertainty." ), all_of=[ r"(compensate|indemnify|reimburse|pay.{0,20}(the\s+)?difference|" r"يعوض|تعويض|يضمن)", r"(difference\s+between|الفرق\s+بين|فرق\s*(?:بين)?)", r"((cost\s+price|purchase\s+price|original\s+price)|" r"سعر\s*(?:التكلفة|الشراء|الأصلي))", r"((the\s+)?sale\s+price|(re)?sale\s*proceeds|" r"سعر\s*(?:البيع|إعادة\s*البيع))", r"(third\s+part(?:y|ies)|طرف\s*ثالث|أطراف\s*ثالثة)", ], applies_to_violation_classes=["Gharar (Excessive Uncertainty)"], max_ml_severity="HIGH", ), ] def scan_hard_negatives(text: str, ml_violation_class: str, ml_severity: str) -> Optional[HardNegativePattern]: """ Returns the first matching hard-negative pattern applicable to this ML result, or None. Applies the violation-class and max-severity guards before even attempting the regex match, so this can never be used to downgrade a violation class or severity it wasn't explicitly built for. """ normalized = normalize_arabic_script(re.sub(r"\s+", " ", text)) ml_rank = _SEVERITY_RANK.get(ml_severity, 0) for pattern in COMPLIANT_HARD_NEGATIVES: if ml_violation_class not in pattern.applies_to_violation_classes: continue if ml_rank > _SEVERITY_RANK.get(pattern.max_ml_severity, 0): continue all_matched = True for rx in pattern.all_of: if not re.search(normalize_arabic_script(rx), normalized, re.IGNORECASE): all_matched = False break if all_matched: return pattern return None def reconcile(text: str, ml_result: dict) -> dict: """ Takes the raw classify() output dict (matching the shape in main.py's classify() — expects ml_result["maci_evaluation"] with violation_class / severity / result / confidence_band, and ml_result["boundary_behavior"] with recommendation) and returns a new dict with those fields reconciled against the structural and hard-negative layers, plus a "reconciliation" block that always explains what happened (even when nothing changed) so the audit trail is honest about which layer produced the final call. """ ev = ml_result["maci_evaluation"] boundary = ml_result["boundary_behavior"] structural_flags: List[StructuralFlag] = scan_structural_patterns(text) reconciliation = { "applied": False, "source": None, "reason": None, "original_ml_violation_class": ev["violation_class"], "original_ml_severity": ev["severity"], "original_ml_result": ev["result"], "original_recommendation": boundary["recommendation"], } # ── Case 1: structural violation overrides, regardless of ML ── if structural_flags: top = max( structural_flags, key=lambda f: _SEVERITY_RANK.get(f.severity, 0), ) ev["violation_class"] = top.name ev["result"] = "FLAGGED" ev["severity"] = top.severity ev["confidence_band"] = "HIGH" ev["model_used"] = ev.get("model_used", "") + " + structural_pattern_layer" boundary["recommendation"] = ( "QUARANTINE" if top.severity == "CRITICAL" else "ESCALATE" ) boundary["review_required"] = True boundary["review_reason"] = f"structural_pattern_match:{top.pattern_id}" ml_result["maqasid_enrichment"] = ml_result.get("maqasid_enrichment", {}) ml_result["maqasid_enrichment"]["aaoifi_standard"] = top.aaoifi_standard ml_result["maqasid_enrichment"]["structural_fix"] = top.explanation reconciliation.update({ "applied": True, "source": "structural_override", "reason": ( f"Deterministic structural pattern {top.pattern_id} " f"({top.name}) matched — this pattern is trusted over " f"the base classifier's verdict for this text, " f"regardless of language." ), "structural_matches": [f.pattern_id for f in structural_flags], }) # ── Case 2: hard-negative downgrade (only if no structural hit) ── else: hn = scan_hard_negatives(text, ev["violation_class"], ev["severity"]) if hn is not None and ev["result"] not in ("AUTHENTIC",): ev["result"] = "REVIEW" boundary["recommendation"] = "REVIEW" boundary["review_required"] = True boundary["review_reason"] = f"hard_negative_pattern_match:{hn.pattern_id}" ml_result["maqasid_enrichment"] = ml_result.get("maqasid_enrichment", {}) ml_result["maqasid_enrichment"]["aaoifi_standard"] = hn.aaoifi_standard ml_result["maqasid_enrichment"]["compliant_alternative"] = ( "No change required — this clause matches a recognized " "AAOIFI-compliant pattern. Flagged for human confirmation, " "not auto-cleared." ) reconciliation.update({ "applied": True, "source": "hard_negative_downgrade", "reason": ( f"Base classifier flagged '{ev['violation_class']}' " f"({reconciliation['original_ml_severity']}), but this text " f"matches known compliant pattern {hn.pattern_id} " f"({hn.name}). Downgraded from an automatic " f"{reconciliation['original_recommendation']} to REVIEW — " f"a human confirms, the system does not self-clear." ), }) ml_result["reconciliation"] = reconciliation return ml_result