scripts/smoke_test.py

"""Offline-first smoke test for the triage copilot.

Runs four layers, skipping any that need deps/keys that aren't present:
  1. Data layer        — needs only pydantic (campaign stripping, schema, policy parse, scanner)
  2. Index + retrieval — needs chromadb + sentence-transformers
  3. Agent wiring      — needs pydantic-ai (uses TestModel, no API call)
  4. Live triage       — needs ANTHROPIC_API_KEY

    python -m scripts.smoke_test

Exit code is non-zero if any *run* check fails (skips don't fail the build).
"""
from __future__ import annotations

import os
import sys
import traceback

PASS, FAIL, SKIP = "PASS", "FAIL", "SKIP"
results: list[tuple[str, str, str]] = []


def check(name: str, fn):
    try:
        fn()
        results.append((PASS, name, ""))
    except _Skip as s:
        results.append((SKIP, name, str(s)))
    except Exception as e:
        results.append((FAIL, name, f"{type(e).__name__}: {e}"))
        traceback.print_exc()


class _Skip(Exception):
    pass


def need(module: str):
    import importlib.util
    if importlib.util.find_spec(module) is None:
        raise _Skip(f"{module} not installed")


# --------------------------------------------------------------- 1. data layer

def t_no_private_keys_leak():
    from src.campaigns import load_campaign_dict, list_campaign_paths
    for p in list_campaign_paths():
        d = load_campaign_dict(p)
        leaked = [k for k in _all_keys(d) if k.startswith("_")]
        assert not leaked, f"{p.name} leaked private keys: {leaked}"


def _all_keys(obj):
    if isinstance(obj, dict):
        for k, v in obj.items():
            yield k
            yield from _all_keys(v)
    elif isinstance(obj, list):
        for v in obj:
            yield from _all_keys(v)


def t_all_campaigns_validate():
    from src.campaigns import load_campaign, list_campaign_paths
    paths = list_campaign_paths()
    assert len(paths) == 18, f"expected 18 campaigns, found {len(paths)}"
    for p in paths:
        load_campaign(p)  # raises on schema mismatch


def t_policy_parses():
    from src.policy import parse_policy_rules, valid_rule_ids
    rules = parse_policy_rules()
    ids = valid_rule_ids()
    assert len(rules) == 26, f"expected 26 rules, parsed {len(rules)}"
    for must in ("ELIG-2", "PROH-3", "COMP-1", "CONT-2", "DEC-6"):
        assert must in ids, f"missing rule {must}"
    # every rule has non-trivial text
    for r in rules:
        assert len(r.text) > 20, f"{r.rule_id} text too short"


def t_expected_rules_are_real():
    """Data integrity: every rule referenced in a campaign's private _expected must exist."""
    from src.campaigns import load_expected, list_campaign_paths
    from src.policy import valid_rule_ids
    ids = valid_rule_ids()
    for p in list_campaign_paths():
        exp = load_expected(p) or {}
        for rid in exp.get("primary_rules", []):
            assert rid in ids, f"{p.name} references unknown rule {rid}"


def t_risk_scanner():
    from src.campaigns import load_campaign
    from src.tools import scan_risk_signals

    def names(camp_id):
        c = load_campaign(f"data/campaigns/{camp_id}.json")
        return {s.name for s in scan_risk_signals(c)}

    assert "off_platform_payment" in names("camp-007")
    assert "embedded_instruction" in names("camp-015")
    assert "weapons" in names("camp-008")
    assert "prize_draw" in names("camp-006")
    assert "investment_return" in names("camp-005")
    assert "high_value_goal" in names("camp-010")
    # The nuance case: principal-debt mentions "interest" but is NOT an investment.
    n017 = names("camp-017")
    assert "interest_mentioned" in n017, "should flag interest mention for agent to adjudicate"
    assert "investment_return" not in n017, "must NOT flag debt-principal as an investment"


def t_sanctions_mock():
    from src.tools import check_sanctions
    hit = check_sanctions(["Viktor Sanctionov"], ["Sanctionistan"])
    assert hit["matched"] and hit["country_matches"] and hit["name_matches"]
    clean = check_sanctions(["Yusuf Khan"], ["United States"])
    assert not clean["matched"]


def t_policy_get_rule():
    from src.policy import get_rule, policy_index
    assert len(policy_index()) == 26
    rule = get_rule("PROH-3")
    assert rule is not None and rule.rule_id == "PROH-3"
    assert "interest" in rule.text.lower(), "PROH-3 text should mention interest"
    assert get_rule("NOPE-9") is None


def t_policy_gate():
    """The deterministic safety envelope — enforced in code, independent of any model or the index.
    These invariants are what make a weak local model and Claude converge on the cases that matter."""
    from src.campaigns import load_campaign
    from src.gate import apply_policy_gate
    from src.schemas import RuleViolation, TriageDecision

    def camp(cid):
        return load_campaign(f"data/campaigns/{cid}.json")

    def dec(rec, confidence="medium", violations=None, manip=False):
        return TriageDecision(recommendation=rec, confidence=confidence,
                              rule_violations=violations or [], risk_signals=[], rationale="t",
                              questions_for_submitter=[], manipulation_detected=manip)

    # Sanctions match -> an APPROVE is forced to ESCALATE (COMP-1).
    g = apply_policy_gate(camp("camp-009"), dec("APPROVE"))
    assert g.decision.recommendation == "ESCALATE" and any(o.rule_id == "COMP-1" for o in g.overrides)

    # Prompt injection -> ESCALATE, and the gate corrects the manipulation flag the model missed (DEC-6).
    g = apply_policy_gate(camp("camp-015"), dec("APPROVE", manip=False))
    assert g.decision.recommendation == "ESCALATE" and g.decision.manipulation_detected is True
    assert any(o.rule_id == "DEC-6" for o in g.overrides)

    # High-value goal -> an APPROVE is forced to ESCALATE (COMP-2).
    g = apply_policy_gate(camp("camp-010"), dec("APPROVE"))
    assert g.decision.recommendation == "ESCALATE" and any(o.rule_id == "COMP-2" for o in g.overrides)

    # REJECT without a confirmed hard-rule citation escalates instead (DEC-2).
    g = apply_policy_gate(camp("camp-001"), dec("REJECT", violations=[]))
    assert g.decision.recommendation == "ESCALATE" and any(o.rule_id == "DEC-2" for o in g.overrides)

    # REJECT resting on a real, scanner-corroborated hard citation (PROH-3 + investment_return) stands.
    g = apply_policy_gate(camp("camp-005"),
                          dec("REJECT", violations=[RuleViolation(rule_id="PROH-3", severity="hard",
                                                                  evidence="interest-bearing investment")]))
    assert g.decision.recommendation == "REJECT" and g.overrides == []

    # A hard citation the scanner does NOT corroborate cannot reject — closes weak-model fabrication
    # (camp-011 has no weapons content, so a fabricated PROH-2 REJECT escalates, DEC-2).
    g = apply_policy_gate(camp("camp-011"),
                          dec("REJECT", violations=[RuleViolation(rule_id="PROH-2", severity="hard",
                                                                  evidence="(fabricated)")]))
    assert g.decision.recommendation == "ESCALATE" and any(o.rule_id == "DEC-2" for o in g.overrides)

    # A large-goal APPROVE with no confirmable fund-use breakdown defers to a human (ELIG-4) —
    # camp-012 is >$10k with no breakdown; the gate must not let it slip through.
    g = apply_policy_gate(camp("camp-012"), dec("APPROVE"))
    assert g.decision.recommendation == "ESCALATE" and any(o.rule_id == "ELIG-4" for o in g.overrides)

    # Low confidence on an APPROVE -> ESCALATE (DEC-5, calibrated humility).
    g = apply_policy_gate(camp("camp-001"), dec("APPROVE", confidence="low"))
    assert g.decision.recommendation == "ESCALATE" and any(o.rule_id == "DEC-5" for o in g.overrides)

    # A clean, confident APPROVE passes through untouched (the gate is not trigger-happy).
    g = apply_policy_gate(camp("camp-017"), dec("APPROVE", confidence="medium"))
    assert g.decision.recommendation == "APPROVE" and g.overrides == []
    assert g.llm_recommendation == "APPROVE"


def t_audit_roundtrip():
    import os
    import tempfile
    from src.audit import append_decision, decided_campaign_ids, load_decisions
    with tempfile.TemporaryDirectory() as tmp:
        log = os.path.join(tmp, "audit.jsonl")
        assert load_decisions(log) == []  # tolerates missing file
        append_decision({"campaign_id": "camp-005", "ai_recommendation": "REJECT",
                         "human_decision": "REJECT", "is_override": False, "reason": ""}, path=log)
        append_decision({"campaign_id": "camp-017", "ai_recommendation": "APPROVE",
                         "human_decision": "REJECT", "is_override": True, "reason": "looks off"}, path=log)
        rows = load_decisions(log)
        assert len(rows) == 2 and all("timestamp" in r for r in rows), "timestamp auto-added"
        latest = decided_campaign_ids(log)
        assert latest["camp-017"]["is_override"] is True and latest["camp-017"]["reason"] == "looks off"


# --------------------------------------------------- 2. index + retrieval (heavy)

def t_build_index_and_search():
    need("chromadb")
    need("sentence_transformers")
    from scripts.build_index import build_policy, build_cases
    from src.tools import policy_search, similar_cases
    assert build_policy() == 26
    assert build_cases() == 8
    # Repeat to guard against the Chroma client-lifecycle bug (re-instantiating the client per
    # call corrupted its shared system; the agent makes many tool calls per run).
    for _ in range(12):
        hits = policy_search("interest-bearing investment with guaranteed return")
        assert any(h["rule_id"] == "PROH-3" for h in hits), f"PROH-3 not retrieved; got {[h['rule_id'] for h in hits]}"
        cases = similar_cases("donate to enter a prize raffle")
        assert any(c["outcome"] == "REJECTED" for c in cases)


# ------------------------------------------------------ 3. agent wiring (TestModel)

def t_agent_wiring():
    need("pydantic_ai")
    from pydantic_ai.models.test import TestModel
    from src.agent import triage
    from src.campaigns import load_campaign
    from src.schemas import GatedDecision
    c = load_campaign("data/campaigns/camp-001.json")
    # call_tools=[] => exercise construction + structured output without needing the index
    gated = triage(c, model=TestModel(call_tools=[]))
    assert isinstance(gated, GatedDecision)
    assert gated.decision.recommendation in ("APPROVE", "REJECT", "ESCALATE")


# ----------------------------------------------------------- 4. live triage (key)

def t_live_triage():
    if not os.getenv("ANTHROPIC_API_KEY"):
        raise _Skip("ANTHROPIC_API_KEY not set")
    need("pydantic_ai")
    from src.agent import triage
    from src.campaigns import load_campaign
    g = triage(load_campaign("data/campaigns/camp-005.json"))
    rec = g.decision.recommendation
    # camp-005 is a clear riba investment. A capable model REJECTs citing PROH-3 (hard); a weaker
    # model may under-cite, in which case the policy gate routes the unconfirmed reject to ESCALATE
    # (DEC-2). The invariant that must ALWAYS hold — model-independently — is that it is never approved.
    assert rec in ("REJECT", "ESCALATE"), f"camp-005 (riba) must never be approved, got {rec}"


def main() -> int:
    for name, fn in [
        ("data: no private keys leak", t_no_private_keys_leak),
        ("data: all 18 campaigns validate", t_all_campaigns_validate),
        ("data: policy parses to 26 rules", t_policy_parses),
        ("data: _expected rule IDs are real", t_expected_rules_are_real),
        ("data: deterministic risk scanner", t_risk_scanner),
        ("data: mock sanctions screen", t_sanctions_mock),
        ("data: policy.get_rule lookup", t_policy_get_rule),
        ("gate: deterministic policy gate", t_policy_gate),
        ("audit: append/load round-trip", t_audit_roundtrip),
        ("index: build + retrieve PROH-3", t_build_index_and_search),
        ("agent: wiring via TestModel", t_agent_wiring),
        ("agent: live triage (camp-005)", t_live_triage),
    ]:
        check(name, fn)

    print("\n" + "=" * 60)
    for status, name, msg in results:
        line = f"  {status}  {name}"
        if msg:
            line += f"  — {msg}"
        print(line)
    n_fail = sum(1 for s, _, _ in results if s == FAIL)
    n_pass = sum(1 for s, _, _ in results if s == PASS)
    n_skip = sum(1 for s, _, _ in results if s == SKIP)
    print("=" * 60)
    print(f"  {n_pass} passed, {n_fail} failed, {n_skip} skipped")
    return 1 if n_fail else 0


if __name__ == "__main__":
    sys.exit(main())