crs.py

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""
CIRCUIT Framework v1.1.0 — Circuit Risk Score (CRS) Calculator
Reference implementation.

Formula:   CRS = Risk Tier × (6 − IMS) × DCW
Max score: 120  (4 × 6 × 5)
Bands:     Green 1–12 | Amber 13–47 | Red 48–96 | Purple 97–120

Rules enforced:
  Rule 3 — Consequence floor: High/Critical tier + DCW ≥ 4 → Red minimum (48)
  Rule 9 — Black-box autonomy limit: Category C + High/Critical + DCW ≥ 3 → blocked

Source of truth: whitepaper.html §4.3.1 and Appendix G
Data file:       ../data/framework.json

Usage (interactive):
    python crs.py

Usage (arguments):
    python crs.py --tier High --ims 2 --dcw Automated --category B
    python crs.py --tier Critical --ims 0 --dcw Catastrophic --category C
    python crs.py --self-test

Importable:
    from crs import calculate_crs, CRSResult
    result = calculate_crs(risk_tier="High", ims=2, dcw="Automated", category="B")
    print(result.band, result.score)
"""

from __future__ import annotations
import argparse
import sys
from dataclasses import dataclass, field
from typing import Optional


# ── Canonical values (whitepaper §4.3.1) ────────────────────────────────────
# Do not change without a formal CIRCUIT Major-revision proposal.

RISK_TIER: dict[str, int] = {
    "Low":      1,
    "Moderate": 2,
    "High":     3,
    "Critical": 4,
}

DCW: dict[str, int] = {
    "Advisory":     1,
    "Recommended":  2,
    "Automated":    3,
    "Irreversible": 4,
    "Catastrophic": 5,
}

BANDS: list[tuple[str, int, int]] = [
    # (name, min_inclusive, max_inclusive)
    ("green",  1,   12),
    ("amber",  13,  47),
    ("red",    48,  96),
    ("purple", 97, 120),
]

BAND_APPROVALS: dict[str, str] = {
    "green":  "Standard change-management approval.",
    "amber":  "AI Governance Committee (AIGC) quarterly review.",
    "red":    "CISO + AIGC sign-off required. ≤ 180 days to reach Amber.",
    "purple": "Not deployable in current configuration.",
}

IMS_CEILING: dict[str, int] = {
    "A": 5,   # Open weights — full internal access
    "B": 3,   # API / foundation model — feature-level max
    "C": 2,   # Embedded vendor AI — behavioral max
}

IMS_LABELS: dict[int, str] = {
    0: "Opaque",
    1: "Behavioral Observability",
    2: "Post-hoc Explainability",
    3: "Feature-level Inspection",
    4: "Circuit-level Inspection",
    5: "Continuous Interpretability",
}

CATEGORY_LABELS: dict[str, str] = {
    "A": "Open weights (self-hosted)",
    "B": "API / foundation model",
    "C": "Embedded vendor AI",
}

# Consequence floor (Rule 3): High/Critical + DCW ≥ 4 → Red minimum
FLOOR_TIER_MIN  = 3   # High
FLOOR_DCW_MIN   = 4   # Irreversible
FLOOR_CRS_MIN   = 48  # Red band minimum

# Rule 9: Category C + High/Critical + DCW ≥ 3 → blocked
RULE9_TIER_MIN  = 3   # High
RULE9_DCW_MIN   = 3   # Automated

FRAMEWORK_VERSION = "1.1.0"


# ── Data classes ─────────────────────────────────────────────────────────────

@dataclass
class CRSResult:
    """Result of a CRS calculation."""
    score: int
    band: str
    risk_tier_label: str
    risk_tier_value: int
    ims: int
    ims_label: str
    ims_ceiling: int
    ims_deficit: int
    dcw_label: str
    dcw_value: int
    category: str
    consequence_floor_applied: bool
    rule9_violation: bool
    ims_ceiling_exceeded: bool
    warnings: list[str] = field(default_factory=list)

    @property
    def approval(self) -> str:
        return BAND_APPROVALS[self.band]

    @property
    def deployable(self) -> bool:
        return not self.rule9_violation and self.band != "purple"


# ── Core calculation ──────────────────────────────────────────────────────────

def assign_band(score: int) -> str:
    """Map a CRS score to its band name. Raises ValueError if out of range."""
    for name, lo, hi in BANDS:
        if lo <= score <= hi:
            return name
    raise ValueError(f"CRS score {score} is outside the valid range 1–120.")


def calculate_crs(
    risk_tier: str,
    ims: int,
    dcw: str,
    category: str = "A",
) -> CRSResult:
    """
    Calculate the Circuit Risk Score (CRS) for a given deployment configuration.

    Args:
        risk_tier: "Low", "Moderate", "High", or "Critical"
        ims:       Interpretability Maturity Score, integer 0–5
        dcw:       Decision Consequence Weight label — "Advisory", "Recommended",
                   "Automated", "Irreversible", or "Catastrophic"
        category:  Model access category — "A", "B", or "C"

    Returns:
        CRSResult dataclass with score, band, flags, and approval requirement.

    Raises:
        ValueError: if any input is invalid.
    """
    # ── Input validation ─────────────────────────────────────────────────────
    if risk_tier not in RISK_TIER:
        raise ValueError(
            f"Invalid risk_tier '{risk_tier}'. "
            f"Must be one of: {', '.join(RISK_TIER)}"
        )
    if not isinstance(ims, int) or not (0 <= ims <= 5):
        raise ValueError(f"IMS must be an integer 0–5. Got: {ims!r}")
    if dcw not in DCW:
        raise ValueError(
            f"Invalid DCW '{dcw}'. "
            f"Must be one of: {', '.join(DCW)}"
        )
    category = category.upper()
    if category not in IMS_CEILING:
        raise ValueError(
            f"Invalid category '{category}'. Must be 'A', 'B', or 'C'."
        )

    tier_val    = RISK_TIER[risk_tier]
    dcw_val     = DCW[dcw]
    ceiling     = IMS_CEILING[category]
    warnings    = []

    # ── IMS ceiling check ────────────────────────────────────────────────────
    ims_ceiling_exceeded = ims > ceiling
    effective_ims = ims
    if ims_ceiling_exceeded:
        warnings.append(
            f"IMS {ims} exceeds Category {category} ceiling of {ceiling}. "
            f"Scoring with IMS {ceiling} (the maximum achievable)."
        )
        effective_ims = ceiling

    deficit = 6 - effective_ims

    # ── Rule 9 check ─────────────────────────────────────────────────────────
    # Category C + High/Critical tier + DCW ≥ Automated → blocked
    # Source: whitepaper §4.4 Rule 9 and EU AI Act Art. 14 crosswalk
    rule9_violation = (
        category == "C"
        and tier_val >= RULE9_TIER_MIN
        and dcw_val >= RULE9_DCW_MIN
    )

    # ── Raw CRS calculation ───────────────────────────────────────────────────
    raw_score = tier_val * deficit * dcw_val

    # ── Consequence floor (Rule 3) ────────────────────────────────────────────
    # High/Critical + DCW ≥ Irreversible → Red minimum regardless of formula
    consequence_floor_applied = (
        tier_val >= FLOOR_TIER_MIN
        and dcw_val >= FLOOR_DCW_MIN
        and raw_score < FLOOR_CRS_MIN
    )
    score = max(raw_score, FLOOR_CRS_MIN) if consequence_floor_applied else raw_score

    # Edge: Rule 9 violation — score is still computed for display, but
    # the deployment is blocked independent of the number.
    if rule9_violation:
        warnings.append(
            f"Rule 9 violation: Category C cannot host {risk_tier}-tier "
            f"workflows with DCW ≥ Automated (DCW = {dcw}). "
            "Reposition to Advisory or Recommended, or re-classify the deployment."
        )

    if consequence_floor_applied:
        warnings.append(
            f"Rule 3 consequence floor applied: {risk_tier} tier + {dcw} DCW "
            f"forces Red minimum. Formula produced {raw_score}; "
            f"floor raises score to {score}."
        )

    band = assign_band(score)

    return CRSResult(
        score=score,
        band=band,
        risk_tier_label=risk_tier,
        risk_tier_value=tier_val,
        ims=effective_ims,
        ims_label=IMS_LABELS[effective_ims],
        ims_ceiling=ceiling,
        ims_deficit=deficit,
        dcw_label=dcw,
        dcw_value=dcw_val,
        category=category,
        consequence_floor_applied=consequence_floor_applied,
        rule9_violation=rule9_violation,
        ims_ceiling_exceeded=ims_ceiling_exceeded,
        warnings=warnings,
    )


# ── Display ───────────────────────────────────────────────────────────────────

BAND_ICONS = {
    "green":  "🟢",
    "amber":  "🟡",
    "red":    "🔴",
    "purple": "🟣",
}

def format_result(r: CRSResult, verbose: bool = True) -> str:
    """Format a CRSResult as a human-readable string."""
    lines = []

    # Header
    icon = BAND_ICONS.get(r.band, "")
    blocked = " [RULE 9 VIOLATION — BLOCKED]" if r.rule9_violation else ""
    lines.append(f"\n{'─' * 60}")
    lines.append(f"  CRS = {r.score}   {icon} {r.band.upper()}{blocked}")
    lines.append(f"{'─' * 60}")

    if verbose:
        lines.append(f"  Formula:  {r.risk_tier_value} × (6 − {r.ims}) × {r.dcw_value}")
        lines.append(f"            = {r.risk_tier_value} × {r.ims_deficit} × {r.dcw_value}")
        lines.append(f"            = {r.score}")
        lines.append("")
        lines.append(f"  Risk Tier:  {r.risk_tier_label} ({r.risk_tier_value})")
        lines.append(f"  IMS:        {r.ims} — {r.ims_label}")
        lines.append(f"  Category:   {r.category} — {CATEGORY_LABELS[r.category]}")
        lines.append(f"              (IMS ceiling: {r.ims_ceiling})")
        lines.append(f"  DCW:        {r.dcw_label} ({r.dcw_value})")
        lines.append("")
        lines.append(f"  Approval:   {r.approval}")
        lines.append("")
        if r.consequence_floor_applied:
            lines.append("  ⚠  Rule 3: Consequence floor applied (High/Critical + DCW ≥ Irreversible)")
        if r.rule9_violation:
            lines.append("  ✗  Rule 9: Category C black-box autonomy limit — deployment blocked")
        if r.ims_ceiling_exceeded:
            lines.append(f"  ⚠  IMS capped at {r.ims_ceiling} (Category {r.category} ceiling)")

    for w in r.warnings:
        lines.append(f"\n  NOTE: {w}")

    lines.append(f"{'─' * 60}\n")
    return "\n".join(lines)


def _two_levers_hint(r: CRSResult) -> str:
    """Show the two paths to a lower score (whitepaper §4.3.4 — two levers)."""
    if r.band == "green":
        return "  ✓ This deployment is Green — no remediation required.\n"

    lines = ["\n  Two paths to a lower score (whitepaper §4.3.4):"]

    # Lever 1: raise IMS
    if r.ims < r.ims_ceiling:
        next_band_ims = r.ims
        for candidate_ims in range(r.ims + 1, r.ims_ceiling + 1):
            candidate = calculate_crs(r.risk_tier_label, candidate_ims, r.dcw_label, r.category)
            if candidate.band != r.band:
                lines.append(
                    f"  1. Raise IMS to {candidate_ims} ({IMS_LABELS[candidate_ims]}) "
                    f"→ CRS drops to {candidate.score} ({candidate.band.upper()})"
                )
                break
        else:
            best = calculate_crs(r.risk_tier_label, r.ims_ceiling, r.dcw_label, r.category)
            lines.append(
                f"  1. Raise IMS to ceiling ({r.ims_ceiling}) "
                f"→ CRS drops to {best.score} ({best.band.upper()})"
            )
    else:
        lines.append(
            f"  1. IMS is already at Category {r.category} ceiling ({r.ims_ceiling}). "
            "Cannot improve via interpretability evidence alone."
        )

    # Lever 2: reduce DCW
    dcw_options = list(DCW.keys())
    current_dcw_idx = dcw_options.index(r.dcw_label)
    if current_dcw_idx > 0:
        for candidate_dcw in dcw_options[:current_dcw_idx]:
            candidate = calculate_crs(r.risk_tier_label, r.ims, candidate_dcw, r.category)
            if candidate.band != r.band and not candidate.rule9_violation:
                lines.append(
                    f"  2. Reduce DCW to {candidate_dcw} "
                    f"→ CRS drops to {candidate.score} ({candidate.band.upper()})"
                )
                break
        else:
            lines.append("  2. No DCW reduction achieves a lower band at current IMS.")
    else:
        lines.append("  2. DCW is already Advisory — cannot reduce further.")

    return "\n".join(lines) + "\n"


# ── Interactive mode ──────────────────────────────────────────────────────────

def _prompt_choice(prompt: str, choices: list[str]) -> str:
    """Prompt the user to choose from a numbered list."""
    print(f"\n  {prompt}")
    for i, c in enumerate(choices, 1):
        print(f"    {i}. {c}")
    while True:
        raw = input("  Enter number: ").strip()
        if raw.isdigit() and 1 <= int(raw) <= len(choices):
            return choices[int(raw) - 1]
        print(f"  Please enter a number between 1 and {len(choices)}.")


def _prompt_int(prompt: str, lo: int, hi: int) -> int:
    """Prompt the user for an integer in a range."""
    while True:
        raw = input(f"\n  {prompt} ({lo}–{hi}): ").strip()
        if raw.isdigit() and lo <= int(raw) <= hi:
            return int(raw)
        print(f"  Please enter an integer between {lo} and {hi}.")


def interactive_mode() -> None:
    """Run the calculator interactively."""
    print("\n╔══════════════════════════════════════════════════╗")
    print("║  CIRCUIT Framework — CRS Calculator              ║")
    print(f"║  v{FRAMEWORK_VERSION}                                       ║")
    print("║  circuitframework.org                            ║")
    print("╚══════════════════════════════════════════════════╝")

    while True:
        print("\n  Enter the deployment details below.")
        print("  (Ctrl-C to exit)\n")

        risk_tier = _prompt_choice("Risk Tier:", list(RISK_TIER.keys()))
        category  = _prompt_choice("Model Category:", ["A — Open weights", "B — API", "C — Embedded vendor"])
        category  = category[0]  # Extract "A", "B", or "C"
        ceiling   = IMS_CEILING[category]
        ims       = _prompt_int(f"IMS (Category {category} ceiling is {ceiling})", 0, 5)
        dcw       = _prompt_choice("Decision Consequence Weight (DCW):", list(DCW.keys()))

        result = calculate_crs(risk_tier, ims, dcw, category)
        print(format_result(result, verbose=True))
        print(_two_levers_hint(result))

        again = input("  Score another? [y/N]: ").strip().lower()
        if again != "y":
            print("\n  Exiting.\n")
            break


# ── Self-test ─────────────────────────────────────────────────────────────────

def self_test() -> None:
    """
    Verify the calculator against worked examples from whitepaper §4.3.3 and
    Appendix D. Exits with code 1 if any assertion fails.
    """
    print(f"Running self-test against whitepaper v{FRAMEWORK_VERSION} examples...\n")
    failures = 0

    cases = [
        # (label, tier, ims, dcw, cat, expected_score, expected_band, floor, rule9)
        # §4.3.3 Example 1 — Category A, SOC agent, High/Advisory/IMS 2
        ("§4.3.3 Ex.1: A/High/IMS2/Advisory",
         "High", 2, "Advisory", "A", 12, "green", False, False),
        # §4.3.3 Example 2 — Category A, malware classifier, Critical/Automated/IMS 2
        ("§4.3.3 Ex.2: A/Critical/IMS2/Automated",
         "Critical", 2, "Automated", "A", 48, "red", False, False),
        # §4.3.3 Example 3 — Category B, SOC triage, High/Advisory/IMS 1
        ("§4.3.3 Ex.3: B/High/IMS1/Advisory",
         "High", 1, "Advisory", "B", 15, "amber", False, False),
        # §4.3.3 Example 3 continued — same model at IMS 3
        ("§4.3.3 Ex.3b: B/High/IMS3/Advisory",
         "High", 3, "Advisory", "B", 9, "green", False, False),
        # Appendix D.1 — Category A, DLP classifier, Critical/Automated/IMS 4
        ("Appendix D.1: A/Critical/IMS4/Automated",
         "Critical", 4, "Automated", "A", 24, "amber", False, False),
        # Appendix D.2 — Category B, SOC triage, High/Advisory/IMS 2
        ("Appendix D.2: B/High/IMS2/Advisory",
         "High", 2, "Advisory", "B", 12, "green", False, False),
        # Appendix D.3 — Category C, M365 Copilot repositioned, Critical/Advisory/IMS 1
        ("Appendix D.3: C/Critical/IMS1/Advisory",
         "Critical", 1, "Advisory", "C", 20, "amber", False, False),
        # Appendix D.3 original — Critical/Catastrophic/IMS 1 → Purple (Rule 9 violation too)
        ("Appendix D.3 original: C/Critical/IMS1/Catastrophic",
         "Critical", 1, "Catastrophic", "C", 100, "purple", False, True),
        # Rule 3 consequence floor — High/Irreversible/IMS 5 formula=12, floor→Red
        ("Rule 3 floor: A/High/IMS5/Irreversible",
         "High", 5, "Irreversible", "A", 48, "red", True, False),
        # Rule 3 consequence floor — Critical/Irreversible/IMS 5 formula=16→floor→Red
        ("Rule 3 floor: A/Critical/IMS5/Irreversible",
         "Critical", 5, "Irreversible", "A", 48, "red", True, False),
        # Rule 9 — Category C + High + Automated
        # Rule 9 blocks independently of score; formula still computes 3×4×3=36, Amber
        # The deployment is blocked via rule9_violation flag, not a score floor
        ("Rule 9: C/High/IMS2/Automated",
         "High", 2, "Automated", "C", 36, "amber", False, True),
        # Max score
        ("Max CRS: A/Critical/IMS0/Catastrophic",
         "Critical", 0, "Catastrophic", "A", 120, "purple", False, False),
        # Min score
        ("Min CRS: A/Low/IMS5/Advisory",
         "Low", 5, "Advisory", "A", 1, "green", False, False),
    ]

    for label, tier, ims, dcw, cat, exp_score, exp_band, exp_floor, exp_rule9 in cases:
        try:
            r = calculate_crs(tier, ims, dcw, cat)
            ok = (r.score == exp_score and r.band == exp_band
                  and r.consequence_floor_applied == exp_floor
                  and r.rule9_violation == exp_rule9)
            status = "PASS" if ok else "FAIL"
            if not ok:
                failures += 1
            print(
                f"  [{status}] {label}\n"
                f"         score={r.score} (expected {exp_score}), "
                f"band={r.band} (expected {exp_band}), "
                f"floor={r.consequence_floor_applied} (expected {exp_floor}), "
                f"rule9={r.rule9_violation} (expected {exp_rule9})"
            )
        except Exception as exc:  # noqa: BLE001
            failures += 1
            print(f"  [ERROR] {label}: {exc}")

    # Verify exactly 34 reachable values (Appendix G)
    reachable = {
        r * (6 - i) * w
        for r in RISK_TIER.values()
        for i in range(6)
        for w in DCW.values()
    }
    expected_count = 34
    count_ok = len(reachable) == expected_count
    status = "PASS" if count_ok else "FAIL"
    if not count_ok:
        failures += 1
    print(
        f"\n  [{status}] Reachable CRS values: {len(reachable)} "
        f"(expected {expected_count})"
    )
    if not count_ok:
        print(f"         Actual reachable set: {sorted(reachable)}")

    print(f"\n{'All tests passed.' if failures == 0 else f'{failures} test(s) FAILED.'}\n")
    sys.exit(0 if failures == 0 else 1)


# ── CLI ───────────────────────────────────────────────────────────────────────

def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="crs.py",
        description=(
            "CIRCUIT Framework CRS Calculator — v"
            + FRAMEWORK_VERSION
            + "\nFormula: CRS = Risk Tier × (6 − IMS) × DCW"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    p.add_argument(
        "--tier", choices=list(RISK_TIER.keys()),
        metavar="TIER",
        help="Risk Tier: Low | Moderate | High | Critical",
    )
    p.add_argument(
        "--ims", type=int, metavar="N",
        help="Interpretability Maturity Score (0–5)",
    )
    p.add_argument(
        "--dcw", choices=list(DCW.keys()),
        metavar="DCW",
        help="Decision Consequence Weight: Advisory | Recommended | Automated | Irreversible | Catastrophic",
    )
    p.add_argument(
        "--category", choices=["A", "B", "C"], default="A",
        metavar="CAT",
        help="Model category: A (open weights) | B (API) | C (embedded vendor). Default: A",
    )
    p.add_argument(
        "--self-test", action="store_true",
        help="Run the built-in self-test against whitepaper worked examples and exit",
    )
    p.add_argument(
        "--quiet", action="store_true",
        help="Output only: score band (e.g. '24 amber') — useful for scripting",
    )
    return p


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()

    if args.self_test:
        self_test()
        return

    # If all three required args are supplied, run non-interactively
    if args.tier and args.ims is not None and args.dcw:
        try:
            result = calculate_crs(args.tier, args.ims, args.dcw, args.category)
        except ValueError as exc:
            print(f"Error: {exc}", file=sys.stderr)
            sys.exit(1)

        if args.quiet:
            print(f"{result.score} {result.band}")
        else:
            print(format_result(result, verbose=True))
            print(_two_levers_hint(result))
        sys.exit(0 if result.deployable else 2)

    # Otherwise, run interactively
    if args.tier or args.ims is not None or args.dcw:
        parser.error("--tier, --ims, and --dcw must all be provided together.")

    try:
        interactive_mode()
    except KeyboardInterrupt:
        print("\n\n  Interrupted.\n")
        sys.exit(0)


if __name__ == "__main__":
    main()