fsci-conformance 0.3.0

#!/usr/bin/env python3
"""SciPy-backed reference oracle capture for FrankenSciPy CASP runtime fixture.

Covers FSCI-P2C-008 runtime CASP (Condition-Aware Solver Portfolio) conformance.
Validates policy decisions, condition-based solver selection, and conformal
calibration against numerical linear algebra conditioning references in SciPy/NumPy.

SciPy has no solver portfolio, so the solver_selection reference is not SciPy's: it is
the argmin, over the general solvers, of the state's column of the measured calibration
committed at artifacts/casp-calibration-solver.json (frankenscipy-7tb8d.2).
"""

from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path
from typing import Any, Dict, List


def _ok(case_id: str, result_kind: str, result: Dict[str, Any]) -> Dict[str, Any]:
    return {
        "case_id": case_id,
        "status": "ok",
        "result_kind": result_kind,
        "result": result,
        "error": None,
    }


def _err(case_id: str, error: str, result_kind: str = "exception") -> Dict[str, Any]:
    return {
        "case_id": case_id,
        "status": "error",
        "result_kind": result_kind,
        "result": {},
        "error": error,
    }


_CALIBRATION = (
    Path(__file__).resolve().parents[3] / "artifacts" / "casp-calibration-solver.json"
)
_STATE_NAMES = {
    "well_conditioned": "WellConditioned",
    "moderate_condition": "Moderate",
    "ill_conditioned": "IllConditioned",
    "near_singular": "NearSingular",
}
# The structured fast paths need structural evidence, which solver_selection cases never carry.
_GENERAL_SOLVERS = {
    "DirectLU": "direct_lu",
    "PivotedQR": "pivoted_qr",
    "SVDFallback": "svd_fallback",
}


def _calibrated_action(cond_state: str) -> str:
    calibration = json.loads(_CALIBRATION.read_text(encoding="utf-8"))
    column = calibration["states"].index(_STATE_NAMES[cond_state])
    losses = {
        action: row[column]
        for action, row in zip(calibration["actions"], calibration["loss_matrix"])
        if action in _GENERAL_SOLVERS
    }
    return _GENERAL_SOLVERS[min(losses, key=losses.__getitem__)]


def _run_case(case: Dict[str, Any], np: Any, scipy: Any) -> Dict[str, Any]:
    case_id = case["case_id"]
    test_kind = case.get("test_kind", "")

    try:
        if test_kind == "policy_decision":
            cond = float(case.get("condition_signal") or 0.0)
            meta = float(case.get("metadata_signal") or 0.0)
            anom = float(case.get("anomaly_signal") or 0.0)

            if meta > 0.5 or cond >= 16.0:
                action = "fail_closed"
            elif cond >= 8.0 or anom > 0.5:
                action = "full_validate"
            else:
                action = "allow"
            return _ok(case_id, "policy_action", {"kind": "policy_action", "action": action})

        if test_kind == "solver_selection":
            cond_state = case.get("condition_state")
            if cond_state not in _STATE_NAMES:
                return _err(case_id, f"unknown condition state: {cond_state}")
            action = _calibrated_action(cond_state)
            return _ok(case_id, "solver_action", {"kind": "solver_action", "action": action})

        if test_kind == "calibrator_drift":
            observations = case.get("observations") or []
            alpha = float(case.get("alpha") or 0.05)
            if len(observations) < 10:
                should_fallback = False
            else:
                violations = sum(1 for obs in observations if obs > 0.0)
                rate = violations / len(observations)
                should_fallback = rate > (alpha * 2.0)
            return _ok(
                case_id,
                "calibrator_fallback",
                {"kind": "calibrator_fallback", "should_fallback": should_fallback},
            )

        return _err(case_id, f"unsupported test_kind: {test_kind}")

    except Exception as exc:
        return _err(case_id, f"{type(exc).__name__}: {exc}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Capture runtime CASP oracle outputs")
    parser.add_argument("--fixture", required=True)
    parser.add_argument("--output", required=True)
    parser.add_argument("--oracle-root", required=False, default="")
    args = parser.parse_args()

    try:
        import numpy as np
        import scipy
        import scipy.linalg
    except ModuleNotFoundError as exc:
        print(str(exc), file=sys.stderr)
        return 2

    fixture_path = Path(args.fixture)
    output_path = Path(args.output)

    try:
        fixture = json.loads(fixture_path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        print(f"Invalid JSON in fixture: {exc}", file=sys.stderr)
        return 1

    case_outputs: List[Dict[str, Any]] = []
    for case in fixture.get("cases", []):
        case_outputs.append(_run_case(case, np=np, scipy=scipy))

    payload = {
        "packet_id": fixture.get("packet_id", "FSCI-P2C-008"),
        "family": fixture.get("family", "runtime_casp"),
        "generated_unix_ms": int(time.time() * 1000),
        "runtime": {
            "python_version": sys.version.split()[0],
            "numpy_version": getattr(np, "__version__", "unknown"),
            "scipy_version": getattr(scipy, "__version__", "unknown"),
        },
        "case_outputs": case_outputs,
    }

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())