src/gate.py

"""The deterministic policy gate — the safety envelope around the LLM.

The agent *recommends*; this gate enforces the policy invariants in code before a human sees the
result. It recomputes the deterministic facts itself (it does NOT trust the model's self-report of
sanctions, manipulation, or risk signals) and reconciles them against the model's recommendation.

Why this exists: without it, the final adjudication is 100% the model's free judgment — so a weak
local model and Claude diverge wildly, and the system is a (well-dressed) LLM wrapper. The gate
makes the safety-critical behavior independent of model IQ: the cases that matter converge.

The envelope has one rule: **the gate may only route toward the human (→ ESCALATE).** It never
produces a more confident decision than the model (never APPROVE→REJECT, never ESCALATE→APPROVE,
never invents a REJECT) and never overrides the human moderator. The AI component can only ever
defer harder to a person — never seize more authority.

Pure functions, no Chroma / no API: the invariants are unit-testable offline and double as the
deterministic layer of the eval harness.
"""
from __future__ import annotations

from . import tools as T
from .policy import valid_rule_ids
from .schemas import Campaign, GatedDecision, GateOverride, RiskSignal, TriageDecision

# High-severity signals route an APPROVE to a human (DEC-3). `embedded_instruction` is excluded
# here because it is handled on the universal manipulation path (DEC-6), and `high_value_goal`
# is cited as COMP-2 (the AML/high-value review rule) rather than the generic DEC-3.
_MANIPULATION_SIGNAL = "embedded_instruction"
_HIGH_VALUE_SIGNAL = "high_value_goal"

# An unverified fund-use plan on a large goal blocks an APPROVE (ELIG-4). This signal fires purely
# on the goal amount (10k < goal <= AML threshold); the scanner cannot confirm a breakdown is
# actually present, so any large-goal APPROVE defers to a human to verify one (calibrated humility).
_BREAKDOWN_SIGNAL = "large_goal_no_breakdown_check"

# Corroboration map: a hard REJECT citing one of these machine-detectable prohibited categories is
# only *confirmed* (DEC-2) if the named deterministic signal independently fired. This closes the
# weak-model failure mode of fabricating a hard category (e.g. citing PROH-2/PROH-4 on a campaign
# with no weapons/prize-draw content) to force a REJECT. Hard rules with NO deterministic detector
# (PROH-1 illegal, PROH-5 hate, PROH-6 adult/intoxicants) are pure-content judgments the scanner
# cannot corroborate — the gate cannot confirm them, so under DEC-2/DEC-5 a citation alone does not
# hold the REJECT and it escalates to a human rather than trusting the model's word.
_HARD_RULE_SIGNALS: dict[str, set[str]] = {
    "PROH-2": {"weapons"},
    "PROH-3": {"investment_return"},
    "PROH-4": {"prize_draw"},
    "PROH-7": {"investment_return"},
    "COMP-3": {"off_platform_payment"},
}


def _has_confirmed_hard_citation(decision: TriageDecision, valid: set[str], signal_names: set[str]) -> bool:
    """A REJECT is only justified by a cited rule that is a *hard* violation, a *real* policy ID,
    AND independently *corroborated* by the deterministic scanner (DEC-2). A hallucinated, soft, or
    uncorroborated citation escalates — suspicion does not reject. Hard rules the scanner cannot
    detect are never deterministically confirmable, so they do not hold a REJECT either."""
    for v in decision.rule_violations:
        if v.severity != "hard" or v.rule_id not in valid:
            continue
        required = _HARD_RULE_SIGNALS.get(v.rule_id)
        if required and (required & signal_names):
            return True
    return False


def _merge_signals(decision_signals: list[RiskSignal], scanned: list[RiskSignal]) -> list[RiskSignal]:
    """Union the model's reported signals with the deterministic scanner's, deduped by name, so the
    moderator always sees ground-truth signals even if the model omitted them."""
    merged = list(decision_signals)
    seen = {s.name for s in merged}
    for s in scanned:
        if s.name not in seen:
            merged.append(s)
            seen.add(s.name)
    return merged


def apply_policy_gate(campaign: Campaign, llm_decision: TriageDecision) -> GatedDecision:
    """Reconcile the model's `TriageDecision` against deterministic policy invariants.

    Returns a `GatedDecision` carrying the final (possibly corrected) decision, the model's original
    recommendation, and an audit trail of every adjustment.
    """
    # Recompute ground truth independently of the model.
    signals = T.scan_risk_signals(campaign)
    sanctions = T.check_sanctions(
        names=[campaign.beneficiary.name, campaign.organizer.name],
        countries=[campaign.beneficiary.country, campaign.organizer.country],
    )
    valid = valid_rule_ids()
    signal_names = {s.name for s in signals}

    original = llm_decision.recommendation

    # The deterministic scanner is authoritative on prompt injection: trust it over the model's flag.
    manipulation = bool(llm_decision.manipulation_detected) or (_MANIPULATION_SIGNAL in signal_names)

    triggers: list[tuple[str, str]] = []

    # --- Universal triggers: the AI must not decide these either way; a human/compliance must. ---
    if sanctions.get("matched"):
        hits = (sanctions.get("country_matches") or []) + (sanctions.get("name_matches") or [])
        triggers.append((
            "COMP-1",
            f"Sanctions/embargo match ({', '.join(hits) or 'matched'}); COMP-1 routes this to "
            "compliance — never auto-decided by the AI.",
        ))
    if manipulation:
        triggers.append((
            "DEC-6",
            "Campaign text contains instructions aimed at the reviewer/AI (prompt injection); "
            "treated as data and escalated, never obeyed.",
        ))

    # --- Block an over-eager APPROVE. ---
    if original == "APPROVE":
        for s in signals:
            if s.severity == "high" and s.name != _MANIPULATION_SIGNAL:
                rule = "COMP-2" if s.name == _HIGH_VALUE_SIGNAL else "DEC-3"
                triggers.append((rule, f"High-severity risk signal '{s.name}': {s.detail}"))
        if _BREAKDOWN_SIGNAL in signal_names:
            triggers.append((
                "ELIG-4",
                "Goal exceeds the $10,000 fund-use-breakdown threshold and a verified breakdown "
                "cannot be confirmed; ELIG-4 routes large-goal approvals to a human to check one "
                "is present (it requires escalation, not rejection).",
            ))
        if llm_decision.confidence == "low":
            triggers.append((
                "DEC-5",
                "Model confidence is low; calibrated humility (DEC-5) routes low-confidence "
                "approvals to a human.",
            ))

    # --- A REJECT must rest on confirmed, corroborated evidence (DEC-2). ---
    if original == "REJECT" and not _has_confirmed_hard_citation(llm_decision, valid, signal_names):
        triggers.append((
            "DEC-2",
            "REJECT requires a confirmed hard-rule citation corroborated by the deterministic "
            "scanner; no cited violation is a valid hard rule backed by independent evidence, so "
            "this escalates (suspicion escalates, it does not reject).",
        ))

    # Build the final decision: merge signals + correct the manipulation flag regardless of outcome.
    final = llm_decision.model_copy(deep=True)
    final.risk_signals = _merge_signals(final.risk_signals, signals)
    final.manipulation_detected = manipulation

    overrides: list[GateOverride] = []
    if triggers and original != "ESCALATE":
        final.recommendation = "ESCALATE"
        overrides = [
            GateOverride(rule_id=rule_id, from_recommendation=original,
                         to_recommendation="ESCALATE", reason=reason)
            for rule_id, reason in triggers
        ]

    return GatedDecision(decision=final, llm_recommendation=original, overrides=overrides)