reconciliation.py

"""
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