Spaces:
Sleeping
Sleeping
| """ | |
| 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, | |
| ) | |
| 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 |