chief-ai-officer-advisor

#!/usr/bin/env python3
"""ai_cost_economics.py — API vs self-hosted inference breakeven analysis.

Stdlib-only. Takes a workload profile and outputs:
  - Monthly API cost at three tiers (frontier-premium, frontier-economy, open-hosted)
  - Monthly self-hosted cost (GPU rental + ops, at chosen model size)
  - Breakeven point: where API and self-hosted cross
  - Sensitivity: low/mid/high GPU rate scenarios
  - Recommended path with explicit caveats

Deterministic logic derived from the profile.

Input schema (JSON):
{
  "workload_name": "Customer support generation",
  "monthly_input_tokens_m": 600,        # millions of input tokens per month
  "monthly_output_tokens_m": 150,
  "quality_tier_required": "frontier-economy",  # frontier-premium | frontier-economy | open-hosted
  "model_size_class_self_host": "70b-class",    # 7b-13b | 70b-class
  "latency_p95_target_ms": 1500,
  "utilization_assumed_pct": 70,                # realistic GPU utilization for self-hosting
  "include_ops_attribution": true               # 30% of an engineer attributed to self-hosted ops
}

Usage:
    python ai_cost_economics.py                       # uses embedded 5M tokens/day sample
    python ai_cost_economics.py path/to/workload.json
    python ai_cost_economics.py workload.json --output json
"""

import argparse
import json
import sys
from typing import Any, Dict, List


SAMPLE: Dict[str, Any] = {
    "workload_name": "B2B SaaS customer-support generation (5M tokens/day)",
    "monthly_input_tokens_m": 600,
    "monthly_output_tokens_m": 150,
    "quality_tier_required": "frontier-economy",
    "model_size_class_self_host": "70b-class",
    "latency_p95_target_ms": 1500,
    "utilization_assumed_pct": 70,
    "include_ops_attribution": True,
}


# 2026 API pricing per million tokens, $USD (input / output)
API_PRICING = {
    "frontier-premium": {"input": 3.00, "output": 15.00, "label": "Claude Sonnet 5 / frontier-mid tier"},
    "frontier-economy": {"input": 1.25, "output": 5.00, "label": "Claude Haiku 4.5 / small-model tier"},
    "open-hosted": {"input": 0.50, "output": 1.50, "label": "Llama 3.1 70B / Qwen 2.5 72B via hosted endpoint"},
}

# GPU spot pricing 2026 ($/hour). Mid-range; varies by provider and commitment.
GPU_PRICING = {
    "A100-spot-low": 1.50,
    "A100-spot-mid": 2.50,
    "A100-spot-high": 3.50,
    "H100-spot-low": 3.50,
    "H100-spot-mid": 5.00,
    "H100-spot-high": 8.00,
}

# Tokens per second per GPU at 70% utilization (rough)
TOKENS_PER_GPU_PER_SEC = {
    "7b-13b": {"A100": 1500, "H100": 3500},
    "70b-class": {"A100": 200, "H100": 600},
}

# Number of GPUs needed for model (minimum, with KV cache)
GPUS_PER_MODEL = {
    "7b-13b": 1,
    "70b-class": 4,   # 70B at FP16 needs ~140GB; 4xA100-40GB or 2xH100-80GB
}

# Engineer fully-loaded cost (annual)
ENGINEER_FULLY_LOADED = 250_000
OPS_ATTRIBUTION_PCT = 0.30   # 30% of an engineer attributed to self-hosted ops


def api_monthly_cost(profile: Dict[str, Any], tier: str) -> float:
    pricing = API_PRICING.get(tier, API_PRICING["frontier-economy"])
    return (
        profile.get("monthly_input_tokens_m", 0) * pricing["input"]
        + profile.get("monthly_output_tokens_m", 0) * pricing["output"]
    )


def self_hosted_monthly_cost(profile: Dict[str, Any], gpu_type: str, gpu_pricing_tier: str) -> Dict[str, Any]:
    """Compute self-hosted monthly cost for given GPU type and pricing tier."""
    model_class = profile.get("model_size_class_self_host", "70b-class")
    utilization = profile.get("utilization_assumed_pct", 70) / 100
    monthly_tokens_total_m = profile.get("monthly_input_tokens_m", 0) + profile.get("monthly_output_tokens_m", 0)
    monthly_tokens_total = monthly_tokens_total_m * 1_000_000

    gpus_needed = GPUS_PER_MODEL[model_class]
    tokens_per_sec_per_gpu = TOKENS_PER_GPU_PER_SEC[model_class][gpu_type]
    effective_tokens_per_sec = gpus_needed * tokens_per_sec_per_gpu * utilization

    # Hours of GPU time needed per month
    seconds_per_month = monthly_tokens_total / effective_tokens_per_sec
    hours_per_month = seconds_per_month / 3600

    # But minimum: GPUs must be warm 24/7 if we want consistent latency
    # So actual hours = max(hours_per_month, 24 * 30 * gpus_needed)
    hours_warm = 24 * 30 * gpus_needed
    hours_billable = max(hours_per_month, hours_warm)

    gpu_pricing_key = f"{gpu_type}-spot-{gpu_pricing_tier}"
    rate = GPU_PRICING[gpu_pricing_key]
    gpu_cost = hours_billable * rate / gpus_needed * gpus_needed  # already per GPU

    ops_cost = (ENGINEER_FULLY_LOADED * OPS_ATTRIBUTION_PCT) / 12 if profile.get("include_ops_attribution", True) else 0

    return {
        "gpu_cost": round(gpu_cost, 0),
        "ops_cost": round(ops_cost, 0),
        "total": round(gpu_cost + ops_cost, 0),
        "hours_warm_required": int(hours_warm),
        "hours_compute_required": int(hours_per_month),
        "gpus_needed": gpus_needed,
        "gpu_rate_per_hr": rate,
    }


def find_breakeven(profile: Dict[str, Any], api_tier: str, gpu_type: str, gpu_pricing_tier: str) -> Dict[str, Any]:
    """Find the monthly token volume where API and self-hosted cost cross."""
    # API cost is linear in tokens; self-hosted has fixed (warm GPU) + linear component
    model_class = profile.get("model_size_class_self_host", "70b-class")
    utilization = profile.get("utilization_assumed_pct", 70) / 100
    gpus_needed = GPUS_PER_MODEL[model_class]
    tokens_per_sec_per_gpu = TOKENS_PER_GPU_PER_SEC[model_class][gpu_type]
    effective_tokens_per_sec = gpus_needed * tokens_per_sec_per_gpu * utilization

    gpu_pricing_key = f"{gpu_type}-spot-{gpu_pricing_tier}"
    rate = GPU_PRICING[gpu_pricing_key]

    # Self-hosted: warm 24/7 fixed cost, plus ops
    monthly_fixed = 24 * 30 * gpus_needed * rate
    ops_cost = (ENGINEER_FULLY_LOADED * OPS_ATTRIBUTION_PCT) / 12 if profile.get("include_ops_attribution", True) else 0
    self_hosted_floor = monthly_fixed + ops_cost  # cost even at zero tokens (because warm)

    # When tokens exceed warm capacity, additional cost is more GPU hours
    # But up to warm capacity, total cost is just monthly_fixed + ops_cost
    warm_capacity_tokens_per_month = effective_tokens_per_sec * 24 * 30 * 3600

    # API cost per million tokens (weighted by I/O ratio)
    monthly_in = profile.get("monthly_input_tokens_m", 1)
    monthly_out = profile.get("monthly_output_tokens_m", 1)
    total_m = monthly_in + monthly_out
    in_ratio = monthly_in / total_m if total_m else 0.8
    out_ratio = monthly_out / total_m if total_m else 0.2

    api_per_m = API_PRICING[api_tier]["input"] * in_ratio + API_PRICING[api_tier]["output"] * out_ratio

    # Breakeven: api_per_m * tokens_m = self_hosted_floor
    if api_per_m > 0:
        breakeven_tokens_m = self_hosted_floor / api_per_m
    else:
        breakeven_tokens_m = None

    return {
        "breakeven_monthly_tokens_m": round(breakeven_tokens_m, 0) if breakeven_tokens_m else None,
        "self_hosted_floor_monthly": round(self_hosted_floor, 0),
        "warm_capacity_monthly_tokens_m": round(warm_capacity_tokens_per_month / 1_000_000, 0),
        "api_per_m_blended": round(api_per_m, 2),
    }


def analyze(profile: Dict[str, Any]) -> Dict[str, Any]:
    api_tier = profile.get("quality_tier_required", "frontier-economy")
    monthly_tokens_total_m = profile.get("monthly_input_tokens_m", 0) + profile.get("monthly_output_tokens_m", 0)

    # API costs at all 3 tiers
    api_costs = {tier: round(api_monthly_cost(profile, tier), 0) for tier in API_PRICING}

    # Self-hosted at chosen GPU type, 3 pricing tiers
    gpu_type = "A100" if profile.get("latency_p95_target_ms", 2000) > 1000 else "H100"
    self_hosted_low = self_hosted_monthly_cost(profile, gpu_type, "low")
    self_hosted_mid = self_hosted_monthly_cost(profile, gpu_type, "mid")
    self_hosted_high = self_hosted_monthly_cost(profile, gpu_type, "high")

    # Breakeven analysis at mid pricing
    breakeven = find_breakeven(profile, api_tier, gpu_type, "mid")

    # Recommendation
    api_chosen_cost = api_costs[api_tier]
    self_hosted_chosen_cost = self_hosted_mid["total"]

    if monthly_tokens_total_m < breakeven["breakeven_monthly_tokens_m"]:
        rec = "API"
        reasoning = (
            f"Current volume ({monthly_tokens_total_m:.0f}M tokens/mo) is BELOW breakeven "
            f"({breakeven['breakeven_monthly_tokens_m']:.0f}M tokens/mo). API tier '{api_tier}' is cheaper "
            f"({_fmt_money(api_chosen_cost)}/mo) than self-hosted "
            f"({_fmt_money(self_hosted_chosen_cost)}/mo at mid GPU rates)."
        )
        caveats = [
            "API costs scale linearly with token volume; revisit when volume doubles",
            "Build multi-vendor abstraction (LiteLLM / OpenRouter / Bifrost) for failover",
            "Pin model IDs; run regression evals on every model upgrade",
        ]
    elif self_hosted_high["total"] < api_chosen_cost:
        rec = "SELF_HOSTED"
        reasoning = (
            f"Current volume ({monthly_tokens_total_m:.0f}M tokens/mo) is well above breakeven. "
            f"Self-hosted at {_fmt_money(self_hosted_chosen_cost)}/mo (mid GPU rates) is cheaper than API "
            f"at {_fmt_money(api_chosen_cost)}/mo across all GPU pricing scenarios."
        )
        caveats = [
            "Quality lags frontier by ~6 months; budget refresh cycle",
            "24/7 on-call required; 30% engineer attribution may underestimate at scale",
            "GPU spot pricing volatile; negotiate reserved capacity at this scale",
            "Eval discipline non-negotiable for self-hosted; without it you cannot detect quality degradation",
        ]
    else:
        rec = "HYBRID"
        reasoning = (
            f"Current volume ({monthly_tokens_total_m:.0f}M tokens/mo) is above breakeven but self-hosted "
            f"cost ({_fmt_money(self_hosted_chosen_cost)}/mo) is close to API ({_fmt_money(api_chosen_cost)}/mo). "
            "Consider hybrid: API for tail / low-volume use cases, self-hosted for high-volume / latency-sensitive paths."
        )
        caveats = [
            "Migration to self-hosted typically takes 3-6 months of engineering time — model in TCO",
            "Hybrid increases operational complexity; ensure routing logic is testable",
            "At this margin, capability differences between API and 70B-class may matter more than cost",
        ]

    return {
        "recommendation": rec,
        "reasoning": reasoning,
        "caveats": caveats,
        "monthly_costs": {
            "api_frontier_premium": api_costs["frontier-premium"],
            "api_frontier_economy": api_costs["frontier-economy"],
            "api_open_hosted": api_costs["open-hosted"],
            "self_hosted_low_gpu_rate": self_hosted_low,
            "self_hosted_mid_gpu_rate": self_hosted_mid,
            "self_hosted_high_gpu_rate": self_hosted_high,
        },
        "breakeven_analysis": breakeven,
        "gpu_type_recommended": gpu_type,
        "current_monthly_tokens_m": monthly_tokens_total_m,
    }


def render_text(result: Dict[str, Any], profile: Dict[str, Any], source: str) -> str:
    lines = []
    lines.append("=" * 72)
    lines.append("AI COST ECONOMICS — API vs SELF-HOSTED")
    lines.append(f"Source: {source}")
    lines.append("=" * 72)
    lines.append("")
    lines.append(f"Workload: {profile.get('workload_name')}")
    lines.append(f"  Volume: {profile.get('monthly_input_tokens_m')}M input + {profile.get('monthly_output_tokens_m')}M output tokens/mo")
    lines.append(f"  Quality tier required: {profile.get('quality_tier_required')}")
    lines.append(f"  Model size for self-host: {profile.get('model_size_class_self_host')}")
    lines.append(f"  Latency p95 target: {profile.get('latency_p95_target_ms')}ms")
    lines.append(f"  Utilization assumed: {profile.get('utilization_assumed_pct')}%")
    lines.append("")
    lines.append("-" * 72)
    lines.append(f"RECOMMENDATION: {result['recommendation']}")
    lines.append("")
    for line in _wrap(result["reasoning"], 2):
        lines.append(line)
    lines.append("")
    lines.append("Caveats:")
    for c in result["caveats"]:
        lines.append(f"  • {c}")
    lines.append("")
    lines.append("-" * 72)
    lines.append("MONTHLY COST COMPARISON:")
    lines.append("")
    mc = result["monthly_costs"]
    lines.append(f"  API frontier-premium:  {_fmt_money(mc['api_frontier_premium']):>15}    ({API_PRICING['frontier-premium']['label']})")
    lines.append(f"  API frontier-economy:  {_fmt_money(mc['api_frontier_economy']):>15}    ({API_PRICING['frontier-economy']['label']})")
    lines.append(f"  API open-hosted:       {_fmt_money(mc['api_open_hosted']):>15}    ({API_PRICING['open-hosted']['label']})")
    lines.append("")
    lines.append(f"  Self-hosted ({result['gpu_type_recommended']}), low GPU rates:   {_fmt_money(mc['self_hosted_low_gpu_rate']['total']):>15}    (GPU @ ${mc['self_hosted_low_gpu_rate']['gpu_rate_per_hr']}/hr × {mc['self_hosted_low_gpu_rate']['gpus_needed']} GPUs)")
    lines.append(f"  Self-hosted ({result['gpu_type_recommended']}), mid GPU rates:   {_fmt_money(mc['self_hosted_mid_gpu_rate']['total']):>15}    (GPU @ ${mc['self_hosted_mid_gpu_rate']['gpu_rate_per_hr']}/hr × {mc['self_hosted_mid_gpu_rate']['gpus_needed']} GPUs)")
    lines.append(f"  Self-hosted ({result['gpu_type_recommended']}), high GPU rates:  {_fmt_money(mc['self_hosted_high_gpu_rate']['total']):>15}    (GPU @ ${mc['self_hosted_high_gpu_rate']['gpu_rate_per_hr']}/hr × {mc['self_hosted_high_gpu_rate']['gpus_needed']} GPUs)")
    lines.append("")
    lines.append(f"  Self-hosted ops attribution: {_fmt_money(mc['self_hosted_mid_gpu_rate']['ops_cost'])}/mo (30% of one engineer)")
    lines.append("")
    lines.append("-" * 72)
    be = result["breakeven_analysis"]
    lines.append("BREAKEVEN ANALYSIS:")
    lines.append("")
    if be["breakeven_monthly_tokens_m"]:
        lines.append(f"  API '{profile.get('quality_tier_required')}' vs self-hosted at mid GPU rates:")
        lines.append(f"    Breakeven: ~{be['breakeven_monthly_tokens_m']:,.0f}M tokens/month")
        lines.append(f"    Current volume: {result['current_monthly_tokens_m']:,.0f}M tokens/month")
        lines.append(f"    Self-hosted floor (warm GPUs + ops, even at zero tokens): {_fmt_money(be['self_hosted_floor_monthly'])}/mo")
        lines.append(f"    Self-hosted warm capacity ceiling: ~{be['warm_capacity_monthly_tokens_m']:,.0f}M tokens/month")
        lines.append(f"    API blended cost: ${be['api_per_m_blended']}/M tokens")
    lines.append("")
    lines.append("-" * 72)
    lines.append("REMINDER: This analysis uses 2026 pricing. Pricing changes; re-run quarterly.")
    lines.append("Migration to self-hosted is 3-6 months of engineering work — model that in your TCO.")
    return "\n".join(lines)


def _fmt_money(amount: float) -> str:
    return f"${amount:,.0f}"


def _wrap(text: str, indent: int, width: int = 70) -> List[str]:
    import textwrap
    return textwrap.wrap(text, width=width, initial_indent=" " * indent, subsequent_indent=" " * indent) or [" " * indent + text]


def main() -> int:
    parser = argparse.ArgumentParser(
        description="API vs self-hosted inference breakeven + sensitivity analysis.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("path", nargs="?", help="Path to workload JSON (uses embedded sample if omitted)")
    parser.add_argument("--output", choices=("text", "json"), default="text", help="Output format")
    args = parser.parse_args()

    if args.path:
        try:
            with open(args.path, "r", encoding="utf-8") as f:
                profile = json.load(f)
            source = args.path
        except (IOError, OSError) as e:
            print(f"error: could not read {args.path}: {e}", file=sys.stderr)
            return 1
        except json.JSONDecodeError as e:
            print(f"error: invalid JSON in {args.path}: {e}", file=sys.stderr)
            return 1
    else:
        profile = SAMPLE
        source = "<embedded sample: 5M tokens/day customer support workload>"

    result = analyze(profile)

    if args.output == "json":
        print(json.dumps({"source": source, "profile": profile, **result}, indent=2))
    else:
        print(render_text(result, profile, source))

    return 0


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

#!/usr/bin/env python3
"""ai_risk_classifier.py — Classify an AI use case under EU AI Act + US state laws.

Stdlib-only. Takes a use case profile and outputs:
  - Risk tier (PROHIBITED / HIGH / LIMITED / MINIMAL) under EU AI Act
  - US state law triggers (NYC LL 144, CO SB 26-189, IL HB 53, CA SB 1001)
  - Industry-specific overlays (FDA, NYDFS, NAIC)
  - Required controls + conformity assessment trigger
  - Citations to specific articles / regulations

NOT legal advice — surfaces classification for qualified AI counsel.

Input schema (JSON):
{
  "use_case": "AI screening of job applications",
  "domain": "employment",                # employment | credit | education | healthcare | critical-infra |
                                          # law-enforcement | biometric | content-moderation | b2b-general |
                                          # consumer-general
  "deploys_in_eu": true,
  "deploys_in_us_states": ["NY", "CO", "IL", "CA"],
  "decisions_affected": "consequential", # consequential | informational | internal-only
  "automation_level": "automated",       # automated | human-in-loop | advisory
  "user_facing": true,
  "biometric_data_processed": false,
  "children_under_16": false
}

Usage:
    python ai_risk_classifier.py                       # uses embedded hiring-AI sample
    python ai_risk_classifier.py path/to/use_case.json
    python ai_risk_classifier.py use_case.json --output json
"""

import argparse
import json
import sys
from typing import Any, Dict, List


SAMPLE: Dict[str, Any] = {
    "use_case": "AI-assisted screening of job applications (resume ranking)",
    "domain": "employment",
    "deploys_in_eu": True,
    "deploys_in_us_states": ["NY", "CO", "IL", "CA"],
    "decisions_affected": "consequential",
    "automation_level": "automated",
    "user_facing": False,
    "biometric_data_processed": False,
    "children_under_16": False,
}


# EU AI Act Annex III "high-risk" domains (Article 6(2))
HIGH_RISK_DOMAINS = {
    "employment",
    "credit",
    "education",
    "critical-infra",
    "law-enforcement",
    "biometric",
    "migration",
    "justice",
    "essential-services",  # insurance, public benefits
}

# EU AI Act Article 5 prohibited practices
PROHIBITED_TRIGGERS = {
    "social-scoring",
    "real-time-biometric-surveillance",
    "subliminal-manipulation",
    "exploitation-of-vulnerability",
    "predictive-policing-from-profiling",
    "emotion-recognition-workplace-or-education",
    "biometric-categorization-by-protected-traits",
}


def classify_eu(profile: Dict[str, Any]) -> Dict[str, Any]:
    """Return EU AI Act classification + reasoning."""
    deploys_eu = profile.get("deploys_in_eu", False)
    if not deploys_eu:
        return {
            "tier": "NOT_APPLICABLE",
            "reasoning": "Does not deploy in EU. EU AI Act not triggered.",
            "obligations": [],
            "citations": [],
        }

    domain = profile.get("domain", "")
    decisions = profile.get("decisions_affected", "informational")
    biometric = profile.get("biometric_data_processed", False)
    automation = profile.get("automation_level", "advisory")
    use_case = profile.get("use_case", "").lower()

    # Article 5 prohibited check (heuristic match)
    for prohibited in PROHIBITED_TRIGGERS:
        if any(kw in use_case for kw in prohibited.split("-")):
            # Conservative: match only if multiple keywords hit
            keywords = prohibited.split("-")
            hits = sum(1 for kw in keywords if kw in use_case)
            if hits >= 2:
                return {
                    "tier": "PROHIBITED",
                    "reasoning": (
                        f"Use case description appears to match Article 5 prohibited practice ({prohibited}). "
                        "Cannot deploy in EU regardless of safeguards. Re-scope the product or exclude EU market."
                    ),
                    "obligations": ["Cease deployment in EU"],
                    "citations": ["EU AI Act Art. 5"],
                }

    # Special prohibited: biometric in public spaces by law enforcement (real-time)
    if biometric and domain == "law-enforcement" and automation == "automated":
        return {
            "tier": "PROHIBITED",
            "reasoning": (
                "Real-time biometric identification by law enforcement in publicly accessible spaces is "
                "Art. 5(1)(h) prohibited (narrow exceptions for serious crimes only)."
            ),
            "obligations": ["Cease deployment unless narrow exception applies, in which case Annex III high-risk obligations also apply"],
            "citations": ["EU AI Act Art. 5(1)(h)"],
        }

    # High-risk Annex III check
    if domain in HIGH_RISK_DOMAINS and decisions == "consequential":
        return {
            "tier": "HIGH",
            "reasoning": (
                f"Annex III high-risk domain ({domain}) with consequential decisions. "
                "Conformity assessment + registration + post-market monitoring required before deployment."
            ),
            "obligations": [
                "Conformity assessment (Art. 43)",
                "Registration in EU AI database (Art. 49)",
                "Risk management system (Art. 9)",
                "Data governance: representative, accurate, complete training data (Art. 10)",
                "Technical documentation maintained throughout lifecycle (Art. 11)",
                "Logging / record-keeping (Art. 12)",
                "Transparency and instructions for use (Art. 13)",
                "Human oversight (Art. 14)",
                "Accuracy, robustness, cybersecurity (Art. 15)",
                "Post-market monitoring + incident reporting (Art. 72)",
            ],
            "citations": ["EU AI Act Art. 6", "Annex III", "Art. 8-15", "Art. 43", "Art. 49", "Art. 72"],
        }

    # Biometric data: special category — usually high-risk
    if biometric:
        return {
            "tier": "HIGH",
            "reasoning": (
                "Biometric data processing triggers Annex III obligations even outside the listed domains "
                "(special category under GDPR Art. 9 + AI Act overlay)."
            ),
            "obligations": [
                "Conformity assessment + Annex III high-risk obligations",
                "GDPR Art. 9(2) explicit consent or other Art. 9 lawful basis",
                "DPIA mandatory (GDPR Art. 35)",
            ],
            "citations": ["EU AI Act Annex III §1", "GDPR Art. 9", "GDPR Art. 35"],
        }

    # Limited risk: chatbots, deepfakes, emotion recognition (outside workplace/edu), generative AI
    if "chatbot" in use_case or "deepfake" in use_case or "image generation" in use_case or "video generation" in use_case:
        return {
            "tier": "LIMITED",
            "reasoning": (
                "Limited risk: transparency obligations apply — users must be informed they are interacting with AI "
                "or that content is AI-generated."
            ),
            "obligations": [
                "Inform users they are interacting with AI (Art. 50(1))",
                "Mark AI-generated / manipulated content (Art. 50(2))",
                "If general-purpose AI model: model card with capabilities, limitations, training-data summary (Art. 53)",
            ],
            "citations": ["EU AI Act Art. 50", "Art. 53"],
        }

    # Minimal risk default
    return {
        "tier": "MINIMAL",
        "reasoning": (
            "Does not fall under prohibited, Annex III high-risk, or limited-risk categories. "
            "No specific AI Act obligations beyond general product safety; voluntary codes of conduct recommended."
        ),
        "obligations": [
            "Voluntary alignment with NIST AI RMF / EU codes of conduct (recommended)",
            "GDPR obligations still apply if personal data is processed",
        ],
        "citations": ["EU AI Act recital 27", "NIST AI RMF 1.0"],
    }


def us_state_triggers(profile: Dict[str, Any]) -> List[Dict[str, str]]:
    """Return list of triggered US state-level obligations."""
    states = set(s.upper() for s in profile.get("deploys_in_us_states", []))
    domain = profile.get("domain", "")
    user_facing = profile.get("user_facing", False)
    triggers = []

    # NYC LL 144 — AEDTs in employment
    if "NY" in states and domain == "employment":
        triggers.append({
            "law": "NYC Local Law 144 (AEDT)",
            "trigger": "Automated Employment Decision Tool used in hiring or promotion for NYC employees",
            "obligations": (
                "Annual independent bias audit; candidate notice (10+ business days before use); "
                "publication of audit summary on company website."
            ),
            "citation": "NYC Local Law 144 of 2021; 6 RCNY § 5-300",
        })

    # Colorado SB 26-189 (effective 2027-01-01; repealed and re-enacted SB 24-205)
    if "CO" in states and domain in {"employment", "credit", "education", "insurance", "essential-services"}:
        triggers.append({
            "law": "Colorado SB 26-189 (replaces the Colorado AI Act, SB 24-205)",
            "trigger": f"Automated decision-making technology in a consequential consumer decision ({domain})",
            "obligations": (
                "Deployer: consumer notice at point of use; plain-language explanation within 30 days "
                "of an adverse outcome; data correction; human review. Developer: technical "
                "documentation and material-update notices to deployers. No impact assessment or "
                "AG filing. Effective 2027-01-01; 60-day cure period."
            ),
            "citation": "Colorado SB 26-189 (2026); CRS § 6-1-1701 et seq.",
        })

    # Illinois HB 53 — AI in employment interviews
    if "IL" in states and domain == "employment":
        triggers.append({
            "law": "Illinois HB 53 (AI Video Interview Act)",
            "trigger": "AI analyzes video interviews of Illinois applicants",
            "obligations": (
                "Candidate notice + consent before recording; explanation of how AI is used; "
                "deletion within 30 days of request; restrictions on sharing data."
            ),
            "citation": "Illinois 820 ILCS 42/",
        })

    # California SB 1001 — Bot disclosure
    if "CA" in states and user_facing:
        triggers.append({
            "law": "California SB 1001 (B.O.T. Act)",
            "trigger": "User-facing AI bot in commercial transactions or California elections",
            "obligations": "Disclose to user that they are interacting with a bot (not a human).",
            "citation": "California Business & Professions Code § 17940",
        })

    # Illinois BIPA — biometric data
    if "IL" in states and profile.get("biometric_data_processed", False):
        triggers.append({
            "law": "Illinois Biometric Information Privacy Act (BIPA)",
            "trigger": "Biometric identifier or biometric information capture",
            "obligations": (
                "Written informed consent; published retention/destruction policy; cannot sell biometric data; "
                "private right of action with statutory damages ($1K-$5K per violation)."
            ),
            "citation": "Illinois 740 ILCS 14/",
        })

    return triggers


def industry_overlays(profile: Dict[str, Any]) -> List[Dict[str, str]]:
    """Return industry-specific regulatory overlays."""
    domain = profile.get("domain", "")
    overlays = []

    if domain == "healthcare":
        overlays.append({
            "framework": "FDA AI/ML guidance + Software as Medical Device (SaMD)",
            "trigger": "AI in clinical decisions, diagnostic, or therapeutic use",
            "obligations": (
                "510(k) or De Novo or PMA pathway depending on risk class; Predetermined Change Control Plan "
                "for adaptive models; Good Machine Learning Practices (GMLP)."
            ),
            "citation": "FDA Guidance on AI/ML SaMD (2023); 21 CFR Part 820",
        })
    elif domain == "credit":
        overlays.append({
            "framework": "ECOA + FCRA + CFPB Circular 2023-03",
            "trigger": "AI used in credit underwriting or adverse action",
            "obligations": (
                "Specific reason for adverse action (not 'algorithm said no'); model risk management "
                "consistent with SR 11-7 if a bank; explainability sufficient for FCRA adverse action notice."
            ),
            "citation": "15 USC §1691 (ECOA); CFPB Circular 2023-03; Fed SR 11-7",
        })
    elif domain == "essential-services":
        overlays.append({
            "framework": "NAIC Model Bulletin on AI in Insurance",
            "trigger": "AI in insurance underwriting, pricing, claims, fraud",
            "obligations": (
                "AI program governance, risk management, third-party AI oversight; "
                "documented testing for unfair discrimination."
            ),
            "citation": "NAIC Model Bulletin on the Use of AI by Insurers (2023)",
        })

    return overlays


def required_controls(profile: Dict[str, Any], eu_classification: Dict[str, Any]) -> List[str]:
    """Return the required-controls checklist based on tier + profile."""
    tier = eu_classification.get("tier", "")
    controls = []

    if tier in ("HIGH", "LIMITED", "MINIMAL"):
        controls.extend([
            "Eval set with documented success criteria before deployment",
            "Monitoring of model output in production (drift, bias, hallucination)",
            "Fallback behavior defined for model failure modes",
            "Human-in-loop review for high-stakes outputs",
        ])

    if tier == "HIGH":
        controls.extend([
            "Conformity assessment completed and documented (EU AI Act Art. 43)",
            "Registration in EU AI database before deployment (Art. 49)",
            "Risk management system documented and maintained (Art. 9)",
            "Training data governance: representativeness, accuracy, bias mitigation (Art. 10)",
            "Technical documentation per Annex IV maintained throughout lifecycle (Art. 11)",
            "Comprehensive logging for traceability (Art. 12)",
            "Human oversight design (e.g., stop button, override capability) (Art. 14)",
            "Post-market monitoring plan + serious incident reporting (Art. 72)",
            "DPIA under GDPR Art. 35 if personal data processed",
        ])

    if tier == "LIMITED":
        controls.extend([
            "User notification: 'You are interacting with AI' or 'This content is AI-generated'",
            "If general-purpose model: publish model card per Art. 53",
        ])

    if profile.get("user_facing"):
        controls.append("Public-facing disclosure of AI usage in customer-facing communications")

    if profile.get("automation_level") == "automated" and profile.get("decisions_affected") == "consequential":
        controls.append("Right-to-explanation / contestation mechanism for affected individuals (GDPR Art. 22)")

    return controls


def analyze(profile: Dict[str, Any]) -> Dict[str, Any]:
    eu = classify_eu(profile)
    us = us_state_triggers(profile)
    overlays = industry_overlays(profile)
    controls = required_controls(profile, eu)

    conformity_required = eu.get("tier") == "HIGH"

    return {
        "eu_classification": eu,
        "us_state_triggers": us,
        "industry_overlays": overlays,
        "required_controls": controls,
        "conformity_assessment_required": conformity_required,
    }


def render_text(result: Dict[str, Any], profile: Dict[str, Any], source: str) -> str:
    lines = []
    lines.append("=" * 72)
    lines.append("AI RISK CLASSIFICATION")
    lines.append(f"Source: {source}")
    lines.append("=" * 72)
    lines.append("")
    lines.append(f"Use case: {profile.get('use_case')}")
    lines.append(f"  Domain: {profile.get('domain')} | Automation: {profile.get('automation_level')} | Decisions: {profile.get('decisions_affected')}")
    lines.append(f"  Deploys in EU: {profile.get('deploys_in_eu')} | US states: {', '.join(profile.get('deploys_in_us_states', []))}")
    lines.append(f"  User-facing: {profile.get('user_facing')} | Biometric: {profile.get('biometric_data_processed')}")
    lines.append("")
    lines.append("-" * 72)
    eu = result["eu_classification"]
    tier_marker = {
        "PROHIBITED": "🔴",
        "HIGH": "🟠",
        "LIMITED": "🟡",
        "MINIMAL": "🟢",
        "NOT_APPLICABLE": "⚪",
    }.get(eu["tier"], "•")
    lines.append(f"EU AI ACT TIER: {tier_marker} {eu['tier']}")
    lines.append("")
    for line in _wrap(eu["reasoning"], 2):
        lines.append(line)
    lines.append("")
    if eu["citations"]:
        lines.append(f"  Citations: {', '.join(eu['citations'])}")
        lines.append("")
    if eu["obligations"]:
        lines.append("  EU obligations:")
        for o in eu["obligations"]:
            lines.append(f"    • {o}")
        lines.append("")
    lines.append("-" * 72)

    lines.append(f"CONFORMITY ASSESSMENT REQUIRED: {'YES' if result['conformity_assessment_required'] else 'no'}")
    lines.append("")
    lines.append("-" * 72)

    us = result["us_state_triggers"]
    if us:
        lines.append(f"US STATE LAW TRIGGERS ({len(us)}):")
        lines.append("")
        for t in us:
            lines.append(f"  • {t['law']}")
            lines.append(f"    Trigger: {t['trigger']}")
            for line in _wrap(t["obligations"], 4):
                lines.append(line)
            lines.append(f"    Citation: {t['citation']}")
            lines.append("")
    else:
        lines.append("US STATE LAW TRIGGERS: none for the listed states + domain.")
        lines.append("")
    lines.append("-" * 72)

    overlays = result["industry_overlays"]
    if overlays:
        lines.append(f"INDUSTRY OVERLAYS ({len(overlays)}):")
        lines.append("")
        for o in overlays:
            lines.append(f"  • {o['framework']}")
            lines.append(f"    Trigger: {o['trigger']}")
            for line in _wrap(o["obligations"], 4):
                lines.append(line)
            lines.append(f"    Citation: {o['citation']}")
            lines.append("")
        lines.append("-" * 72)

    lines.append(f"REQUIRED CONTROLS ({len(result['required_controls'])}):")
    for c in result["required_controls"]:
        lines.append(f"  ☐ {c}")
    lines.append("")
    lines.append("-" * 72)
    lines.append("REMINDER: This is triage, not legal advice. EU AI Act conformity assessment requires qualified")
    lines.append("AI counsel and may require Notified Body involvement. Re-run quarterly as regulations evolve.")
    return "\n".join(lines)


def _wrap(text: str, indent: int, width: int = 70) -> List[str]:
    import textwrap
    return textwrap.wrap(text, width=width, initial_indent=" " * indent, subsequent_indent=" " * indent) or [" " * indent + text]


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Classify an AI use case under EU AI Act + US state laws.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("path", nargs="?", help="Path to use_case JSON (uses embedded sample if omitted)")
    parser.add_argument("--output", choices=("text", "json"), default="text", help="Output format")
    args = parser.parse_args()

    if args.path:
        try:
            with open(args.path, "r", encoding="utf-8") as f:
                profile = json.load(f)
            source = args.path
        except (IOError, OSError) as e:
            print(f"error: could not read {args.path}: {e}", file=sys.stderr)
            return 1
        except json.JSONDecodeError as e:
            print(f"error: invalid JSON in {args.path}: {e}", file=sys.stder… [409 more characters not shown]

#!/usr/bin/env python3
"""model_buildvsbuy_calculator.py — Decide API vs fine-tune vs build for a use case.

Stdlib-only. Takes a use case profile and outputs:
  - Recommendation (API / FINE_TUNE / BUILD) with reasoning
  - 3-year TCO comparison across all 3 paths
  - Breakeven analysis (where API stops being cheapest)
  - Failure modes for the chosen path

Deterministic logic derived from the profile.

Input schema (JSON):
{
  "use_case": "Customer support response generation",
  "expected_qps": 5,                           # queries per second peak
  "monthly_volume_queries": 4000000,           # queries per month
  "avg_tokens_in": 800,
  "avg_tokens_out": 200,
  "latency_budget_ms": 2000,
  "accuracy_required": "frontier",             # frontier | high | acceptable
  "domain_specific": false,                    # need specific vocabulary / format / behavior
  "data_for_finetune_available": false,        # do we have labeled data for fine-tune?
  "team_ml_capacity_engineers": 1,
  "compliance_requires_self_host": false       # data residency / sovereignty constraint
}

Usage:
    python model_buildvsbuy_calculator.py                       # uses embedded customer-support sample
    python model_buildvsbuy_calculator.py path/to/use_case.json
    python model_buildvsbuy_calculator.py use_case.json --output json
"""

import argparse
import json
import sys
from typing import Any, Dict, List, Tuple


SAMPLE: Dict[str, Any] = {
    "use_case": "Customer support response generation (B2B SaaS)",
    "expected_qps": 5,
    "monthly_volume_queries": 4_000_000,
    "avg_tokens_in": 800,
    "avg_tokens_out": 200,
    "latency_budget_ms": 2000,
    "accuracy_required": "high",
    "domain_specific": True,
    "data_for_finetune_available": False,
    "team_ml_capacity_engineers": 1,
    "compliance_requires_self_host": False,
}


# 2026 API pricing per million tokens, $USD (input / output). These are illustrative;
# real pricing changes; rerun this calculator quarterly.
API_PRICING = {
    "frontier-premium": {"input": 3.00, "output": 15.00, "label": "Claude Sonnet 5 / frontier-mid tier"},
    "frontier-economy": {"input": 1.25, "output": 5.00, "label": "Claude Haiku 4.5 / small-model tier"},
    "open-router-hosted": {"input": 0.50, "output": 1.50, "label": "Llama 3.1 70B / Qwen 2.5 72B via hosted endpoint"},
}

# Fine-tune cost (one-time + ongoing)
FINETUNE_ONE_TIME = 25_000          # data prep + initial training + eval harness
FINETUNE_ANNUAL_RETRAIN = 15_000    # quarterly retraining + ops
FINETUNE_INFERENCE_PER_M = 0.40     # cost per M tokens at moderate scale on hosted endpoint

# Self-hosted inference cost (per million tokens, including GPU + ops at 70% utilization)
SELF_HOSTED_PER_M = {
    "7b-13b": 0.15,
    "70b-class": 1.50,
    "frontier-class": 12.00,   # very expensive without massive scale; included for completeness
}

# Build-from-scratch cost (one-time + ongoing) — illustrative; usually NOT recommended
BUILD_FROM_SCRATCH_ONE_TIME = 8_000_000
BUILD_FROM_SCRATCH_ANNUAL = 3_000_000


def compute_api_cost_3yr(profile: Dict[str, Any], tier: str) -> float:
    """3-year API cost given workload."""
    monthly_queries = profile.get("monthly_volume_queries", 0)
    tokens_in = profile.get("avg_tokens_in", 0)
    tokens_out = profile.get("avg_tokens_out", 0)

    monthly_input_tokens_m = (monthly_queries * tokens_in) / 1_000_000
    monthly_output_tokens_m = (monthly_queries * tokens_out) / 1_000_000

    pricing = API_PRICING.get(tier, API_PRICING["frontier-premium"])
    monthly_cost = (
        monthly_input_tokens_m * pricing["input"]
        + monthly_output_tokens_m * pricing["output"]
    )
    return monthly_cost * 36  # 3 years


def compute_finetune_cost_3yr(profile: Dict[str, Any]) -> float:
    monthly_queries = profile.get("monthly_volume_queries", 0)
    tokens_total = profile.get("avg_tokens_in", 0) + profile.get("avg_tokens_out", 0)
    monthly_tokens_m = (monthly_queries * tokens_total) / 1_000_000

    monthly_inference = monthly_tokens_m * FINETUNE_INFERENCE_PER_M
    annual_inference = monthly_inference * 12
    return FINETUNE_ONE_TIME + (annual_inference + FINETUNE_ANNUAL_RETRAIN) * 3


def compute_self_hosted_cost_3yr(profile: Dict[str, Any], model_class: str) -> float:
    """3-year self-hosted cost including GPU + ops."""
    monthly_queries = profile.get("monthly_volume_queries", 0)
    tokens_total = profile.get("avg_tokens_in", 0) + profile.get("avg_tokens_out", 0)
    monthly_tokens_m = (monthly_queries * tokens_total) / 1_000_000

    per_m = SELF_HOSTED_PER_M.get(model_class, SELF_HOSTED_PER_M["70b-class"])
    monthly_inference = monthly_tokens_m * per_m

    # Add fixed ops cost: 1 engineer * 30% load * fully-loaded $250K/yr = $75K/yr ops attribution
    annual_ops = 75_000
    return (monthly_inference * 36) + (annual_ops * 3)


def compute_build_cost_3yr() -> float:
    return BUILD_FROM_SCRATCH_ONE_TIME + (BUILD_FROM_SCRATCH_ANNUAL * 3)


def pick_recommendation(profile: Dict[str, Any], costs: Dict[str, float]) -> Tuple[str, str, List[str]]:
    """Pick API / FINE_TUNE / BUILD with reasoning and failure modes."""
    accuracy = profile.get("accuracy_required", "high")
    domain_specific = profile.get("domain_specific", False)
    finetune_data = profile.get("data_for_finetune_available", False)
    ml_capacity = profile.get("team_ml_capacity_engineers", 0)
    self_host_required = profile.get("compliance_requires_self_host", False)
    latency_ms = profile.get("latency_budget_ms", 2000)
    monthly_q = profile.get("monthly_volume_queries", 0)

    # Special case: compliance forces self-host
    if self_host_required:
        return (
            "FINE_TUNE",
            (
                "Compliance / data residency forces self-host. Fine-tune a 70B-class open model "
                f"({_fmt_money(costs['finetune_3yr'])}/3yr) rather than build from scratch "
                f"({_fmt_money(costs['build_3yr'])}/3yr) — the gap is two orders of magnitude with "
                "comparable quality for most use cases."
            ),
            [
                "Quality lags frontier by ~6 months; budget for refresh every 12-18mo",
                "Self-hosting requires 24/7 on-call; budget 30%+ of an engineer FTE",
                "Eval discipline becomes non-negotiable; without an eval set you cannot tell when retraining is needed",
            ],
        )

    # Build from scratch — almost never
    if accuracy == "frontier" and monthly_q > 1_000_000_000 and ml_capacity >= 20:
        return (
            "BUILD",
            (
                "Edge case where frontier accuracy + extreme volume + large ML team justify pre-training. "
                "Cost still extreme. Most companies here are foundation-model startups, not application companies."
            ),
            [
                "By the time you ship, frontier models have caught up — sunk cost risk",
                "Requires sustained $50M+ investment over 18+ months",
                "Unless model IS your product, do not build",
            ],
        )

    # Fine-tune cases
    if domain_specific and finetune_data and ml_capacity >= 2:
        return (
            "FINE_TUNE",
            (
                "Domain-specific behavior + labeled data + ML engineering capacity available. "
                f"Fine-tune cost ({_fmt_money(costs['finetune_3yr'])}) competes with API at this volume."
            ),
            [
                "Fine-tuned model lags frontier by ~6 months; quality drift is inevitable",
                "Retraining cadence (quarterly typical) is a recurring engineering cost",
                "Without eval set, fine-tune drift is invisible until customer complains",
            ],
        )

    # Latency-driven fine-tune (sub-500ms with 70B-class)
    if latency_ms < 500 and monthly_q > 1_000_000:
        return (
            "FINE_TUNE",
            (
                f"Latency budget {latency_ms}ms below frontier-API median (~600-1500ms). "
                "Fine-tuned 70B-class on dedicated infra is the path to sub-500ms at scale."
            ),
            [
                "Sub-500ms requires GPU co-location and warm pools (idle time penalty)",
                "Quality must be re-verified at every model swap",
                "Streaming responses can buy headroom on latency budget; consider before committing to fine-tune",
            ],
        )

    # Default to API for everything else
    economy_acceptable = accuracy in ("acceptable", "high")
    if economy_acceptable and costs["api_economy_3yr"] < costs["finetune_3yr"]:
        return (
            "API",
            (
                f"Frontier-economy API tier ({API_PRICING['frontier-economy']['label']}) at "
                f"{_fmt_money(costs['api_economy_3yr'])}/3yr beats fine-tune ({_fmt_money(costs['finetune_3yr'])}/3yr). "
                "Iterate on prompt engineering and eval discipline before committing to fine-tune."
            ),
            [
                "Vendor lock-in: build abstraction layer (LiteLLM, OpenRouter, Bifrost) for multi-vendor failover",
                "Capability drift between model versions: pin model IDs and run regression evals on upgrades",
                "Rate limits at QPS spikes: confirm Tier-4+ pricing with provider",
            ],
        )
    return (
        "API",
        (
            f"Frontier-premium API at {_fmt_money(costs['api_premium_3yr'])}/3yr is the right starting point. "
            "Revisit fine-tune at ≥10M queries/month OR domain-specific behavior the API can't be prompted into."
        ),
        [
            "Vendor lock-in: build abstraction layer for multi-vendor failover",
            "Capability drift between model versions; pin model IDs",
            "Rate limits at QPS spikes; confirm pricing tier with provider",
        ],
    )


def analyze(profile: Dict[str, Any]) -> Dict[str, Any]:
    costs = {
        "api_premium_3yr": compute_api_cost_3yr(profile, "frontier-premium"),
        "api_economy_3yr": compute_api_cost_3yr(profile, "frontier-economy"),
        "api_open_hosted_3yr": compute_api_cost_3yr(profile, "open-router-hosted"),
        "finetune_3yr": compute_finetune_cost_3yr(profile),
        "self_hosted_70b_3yr": compute_self_hosted_cost_3yr(profile, "70b-class"),
        "build_3yr": compute_build_cost_3yr(),
    }

    recommendation, reasoning, failure_modes = pick_recommendation(profile, costs)

    # Compute breakeven volume where API and fine-tune cross
    monthly_q = profile.get("monthly_volume_queries", 1)
    tokens_per_q = profile.get("avg_tokens_in", 0) + profile.get("avg_tokens_out", 0)
    annual_q = monthly_q * 12

    # Find breakeven where API economy total == fine-tune total over 3 years
    if tokens_per_q and annual_q:
        api_economy_per_query = costs["api_economy_3yr"] / (annual_q * 3) if annual_q else 0
        # finetune_cost = ONE_TIME + (queries * tokens * inference_per_m / 1M + ANNUAL_RETRAIN) * 3
        # Solve for queries where api_cost == finetune_cost
        # api_economy_per_query * Q = FINETUNE_ONE_TIME + (Q * tokens_per_q * FINETUNE_INFERENCE_PER_M / 1M + RETRAIN) * 3
        # api_economy_per_query * Q - 3 * Q * tokens_per_q * FINETUNE_INFERENCE_PER_M / 1M = FINETUNE_ONE_TIME + 3 * RETRAIN
        # Q * (api_economy_per_query - 3 * tokens_per_q * FINETUNE_INFERENCE_PER_M / 1M) = ONE_TIME + 3 * RETRAIN
        coefficient = (
            api_economy_per_query
            - 3 * tokens_per_q * FINETUNE_INFERENCE_PER_M / 1_000_000
        )
        rhs = FINETUNE_ONE_TIME + 3 * FINETUNE_ANNUAL_RETRAIN
        breakeven_3yr_queries = int(rhs / coefficient) if coefficient > 0 else None
        breakeven_monthly_queries = int(breakeven_3yr_queries / 36) if breakeven_3yr_queries else None
    else:
        breakeven_monthly_queries = None

    return {
        "recommendation": recommendation,
        "reasoning": reasoning,
        "failure_modes": failure_modes,
        "costs_3yr_usd": {k: round(v, 0) for k, v in costs.items()},
        "breakeven_monthly_queries_api_vs_finetune": breakeven_monthly_queries,
        "current_monthly_volume": profile.get("monthly_volume_queries", 0),
    }


def render_text(result: Dict[str, Any], profile: Dict[str, Any], source: str) -> str:
    lines = []
    lines.append("=" * 72)
    lines.append("MODEL BUILD-VS-BUY ANALYSIS")
    lines.append(f"Source: {source}")
    lines.append("=" * 72)
    lines.append("")
    lines.append(f"Use case: {profile.get('use_case')}")
    lines.append(f"  Volume: {profile.get('monthly_volume_queries'):,} queries/mo @ {profile.get('expected_qps')} QPS peak")
    lines.append(f"  Tokens: {profile.get('avg_tokens_in')} in / {profile.get('avg_tokens_out')} out per query")
    lines.append(f"  Latency budget: {profile.get('latency_budget_ms')}ms | Accuracy: {profile.get('accuracy_required')}")
    lines.append(f"  Domain-specific: {profile.get('domain_specific')} | Fine-tune data available: {profile.get('data_for_finetune_available')}")
    lines.append(f"  ML capacity: {profile.get('team_ml_capacity_engineers')} engineers | Compliance forces self-host: {profile.get('compliance_requires_self_host')}")
    lines.append("")
    lines.append("-" * 72)
    lines.append(f"RECOMMENDATION: {result['recommendation']}")
    lines.append("")
    for line in _wrap(result["reasoning"], 2):
        lines.append(line)
    lines.append("")
    lines.append("Failure modes to plan for:")
    for fm in result["failure_modes"]:
        lines.append(f"  • {fm}")
    lines.append("")
    lines.append("-" * 72)
    lines.append("3-YEAR TCO COMPARISON ($ USD):")
    lines.append("")
    costs = result["costs_3yr_usd"]
    lines.append(f"  API (frontier-premium, {API_PRICING['frontier-premium']['label']}):  {_fmt_money(costs['api_premium_3yr']):>15}")
    lines.append(f"  API (frontier-economy, {API_PRICING['frontier-economy']['label']}):   {_fmt_money(costs['api_economy_3yr']):>15}")
    lines.append(f"  API (open-router-hosted, {API_PRICING['open-router-hosted']['label']}): {_fmt_money(costs['api_open_hosted_3yr']):>15}")
    lines.append(f"  Fine-tune (70B-class, hosted inference):                                  {_fmt_money(costs['finetune_3yr']):>15}")
    lines.append(f"  Self-hosted (70B-class on rented H100/A100):                              {_fmt_money(costs['self_hosted_70b_3yr']):>15}")
    lines.append(f"  Build from scratch (pre-train + ops):                                     {_fmt_money(costs['build_3yr']):>15}")
    lines.append("")
    if result["breakeven_monthly_queries_api_vs_finetune"]:
        lines.append(f"Breakeven: API (economy) vs fine-tune crosses at ~{result['breakeven_monthly_queries_api_vs_finetune']:,} queries/month")
        if result["current_monthly_volume"] < result["breakeven_monthly_queries_api_vs_finetune"]:
            lines.append(f"  Current volume ({result['current_monthly_volume']:,}/mo) is BELOW breakeven → API still cheaper.")
        else:
            lines.append(f"  Current volume ({result['current_monthly_volume']:,}/mo) is ABOVE breakeven → fine-tune economics favorable.")
    lines.append("")
    lines.append("-" * 72)
    lines.append("REMINDER: TCO does not capture quality cost. Fine-tune quality lags frontier by ~6 months;")
    lines.append("self-hosted requires eval discipline you may not have. Re-run quarterly with updated pricing.")
    return "\n".join(lines)


def _fmt_money(amount: float) -> str:
    return f"${amount:,.0f}"


def _wrap(text: str, indent: int, width: int = 70) -> List[str]:
    import textwrap
    return textwrap.wrap(text, width=width, initial_indent=" " * indent, subsequent_indent=" " * indent) or [" " * indent + text]


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Decide API vs fine-tune vs build with 3-year TCO comparison.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("path", nargs="?", help="Path to use_case JSON (uses embedded sample if omitted)")
    parser.add_argument("--output", choices=("text", "json"), default="text", help="Output format")
    args = parser.parse_args()

    if args.path:
        try:
            with open(args.path, "r", encoding="utf-8") as f:
                profile = json.load(f)
            source = args.path
        except (IOError, OSError) as e:
            print(f"error: could not read {args.path}: {e}", file=sys.stderr)
            return 1
        except json.JSONDecodeError as e:
            print(f"error: invalid JSON in {args.path}: {e}", file=sys.stderr)
            return 1
    else:
        profile = SAMPLE
        source = "<embedded sample: B2B SaaS customer-support generation, 4M queries/mo>"

    result = analyze(profile)

    if args.output == "json":
        print(json.dumps({"source": source, "profile": profile, **result}, indent=2))
    else:
        print(render_text(result, profile, source))

    return 0


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