stem_ai/policy_intent.py

from __future__ import annotations

from copy import deepcopy
from typing import Any

from .calibration_profile import load_calibration_profile, validate_profile


INTENT_KEYS = (
    "clinical_strictness",
    "code_integrity_priority",
    "reproducibility_priority",
    "structured_limitations_requirement",
)


def validate_intent_answers(answers: dict[str, int]) -> None:
    missing = [key for key in INTENT_KEYS if key not in answers]
    if missing:
        raise ValueError(f"Missing intent answers: {', '.join(missing)}")
    for key in INTENT_KEYS:
        value = answers[key]
        if not isinstance(value, int) or value < 1 or value > 5:
            raise ValueError(f"{key} must be an integer in 1..5")


def derive_policy_intent(
    answers: dict[str, int],
    *,
    baseline_profile_name: str = "default",
) -> dict[str, Any]:
    validate_intent_answers(answers)
    baseline = load_calibration_profile(baseline_profile_name)
    clinical = answers["clinical_strictness"]
    code_priority = answers["code_integrity_priority"]
    reproducibility = answers["reproducibility_priority"]
    structured_limits = answers["structured_limitations_requirement"]

    derived: dict[str, Any] = {
        "baseline_profile": baseline_profile_name,
        "answers": deepcopy(answers),
        "rule_mode": "top_down_first_match",
        "outcome_type": "",
        "recommended_profile": "",
        "triggered_rules": [],
        "notes": [],
        "preview_only_deltas": {},
    }

    if clinical >= 4 and reproducibility <= 3:
        derived["outcome_type"] = "named_profile"
        derived["recommended_profile"] = "strict_clinical_adjacency"
        derived["triggered_rules"].append("clinical_strictness>=4 and reproducibility_priority<=3")
        derived["notes"].append("Strong clinical strictness maps to the existing strict clinical-adjacency profile.")
        return derived

    if baseline_profile_name == "default" and all(2 <= answers[key] <= 3 for key in INTENT_KEYS):
        derived["outcome_type"] = "default_match"
        derived["recommended_profile"] = "default"
        derived["triggered_rules"].append("all_four_values_in_2_to_3_range")
        derived["notes"].append("The default profile already matches the stated posture closely enough.")
        return derived

    preview_deltas: dict[str, Any] = {}
    if clinical >= 4:
        preview_deltas["clinical_policy"] = {
            "ca_no_disclaimer_cap": min(baseline["clinical_policy"]["ca_no_disclaimer_cap"], 60),
            "t0_hard_floor_cap": min(baseline["clinical_policy"]["t0_hard_floor_cap"], 35),
        }
        derived["notes"].append("Clinical strictness requests a stricter cap posture in preview-only mode.")
    if code_priority >= 4:
        preview_deltas["weights"] = {
            "stage_1_percent": 35,
            "stage_2r_percent": 20,
            "stage_3_percent": 45,
        }
        derived["notes"].append("Code-integrity priority shifts 5 points from Stage 1 to Stage 3 in preview-only mode.")
    if reproducibility >= 4:
        preview_deltas["stage_4_policy"] = {"emphasis": "stronger_than_baseline"}
        derived["notes"].append("Reproducibility priority raises Stage 4 emphasis, but does not change the formal score in the current engine.")
    if structured_limits >= 4:
        preview_deltas["stage_3_policy"] = {"b2_partial_credit_mode": "structured_boundary_required"}
        derived["notes"].append("Structured limitations requirement keeps the stricter B2 posture active.")
    if not preview_deltas:
        derived["notes"].append("No named profile rule matched and no explicit bounded delta was activated.")

    derived["outcome_type"] = "preview_only"
    derived["recommended_profile"] = "preview_only"
    derived["triggered_rules"].append("fallback_preview_only")
    derived["preview_only_deltas"] = preview_deltas
    return derived


def simulate_policy_outcome(
    result: dict[str, Any],
    derived: dict[str, Any] | None,
    *,
    baseline_profile_name: str = "default",
    external_profile: dict[str, Any] | None = None,
) -> dict[str, Any]:
    baseline_profile = load_calibration_profile(baseline_profile_name)
    effective_profile = deepcopy(baseline_profile)
    notes: list[str]
    baseline_stage_4_emphasis = baseline_profile.get("stage_4_policy", {}).get("emphasis", "unknown")

    if external_profile is not None:
        effective_profile = deepcopy(external_profile)
        outcome_type = "external_profile_file"
        notes = [
            f"Local profile file used for simulation only: {effective_profile.get('policy_path', '(unknown path)')}",
            "This simulation does not register or promote the local file on the authoritative score path.",
        ]
    else:
        if derived is None:
            raise ValueError("derived policy intent is required unless an external profile is supplied")
        outcome_type = derived["outcome_type"]
        recommended_profile = derived["recommended_profile"]
        notes = list(derived.get("notes", []))

        if outcome_type == "named_profile" and recommended_profile != baseline_profile_name:
            effective_profile = load_calibration_profile(recommended_profile)
        elif outcome_type == "preview_only":
            _apply_preview_deltas(effective_profile, derived.get("preview_only_deltas", {}))
            validate_profile(effective_profile)

    raw_score = _simulate_weighted_raw_score(result, effective_profile)
    baseline_score_cap = _baseline_score_cap(result)
    score_cap = _simulate_score_cap(result, effective_profile)
    final_score = min(raw_score, score_cap) if score_cap is not None else raw_score
    tier = _tier_from_policy(final_score, effective_profile["tier_policy"])
    effective_stage_4_emphasis = effective_profile.get("stage_4_policy", {}).get("emphasis", "unknown")

    baseline_raw = int(result["score"]["raw_score_before_floor"])
    baseline_final = int(result["score"]["final_score"])
    replication_posture_changed = effective_stage_4_emphasis != baseline_stage_4_emphasis
    if replication_posture_changed:
        notes.append(
            "Stage 4 replication posture changed: "
            f"{baseline_stage_4_emphasis} -> {effective_stage_4_emphasis}."
        )
    if replication_posture_changed and final_score == baseline_final:
        notes.append(
            "Formal score remained unchanged because Stage 4 is still a separate "
            "replication lane in 1.8.0."
        )

    simulation = {
        "baseline_profile": baseline_profile_name,
        "effective_profile": effective_profile["profile_name"],
        "effective_policy_version": effective_profile["policy_version"],
        "effective_profile_status": effective_profile["profile_status"],
        "effective_profile_read_mode": effective_profile["profile_read_mode"],
        "effective_policy_sha256": effective_profile["policy_sha256"],
        "effective_profile_source": "local_file" if external_profile is not None else "named_profile",
        "effective_profile_path": effective_profile.get("policy_path"),
        "outcome_type": outcome_type,
        "baseline_stage_4_emphasis": baseline_stage_4_emphasis,
        "effective_stage_4_emphasis": effective_stage_4_emphasis,
        "replication_posture_changed": replication_posture_changed,
        "baseline_score_cap": baseline_score_cap,
        "raw_score_before_cap": raw_score,
        "score_cap": score_cap,
        "score_cap_changed": score_cap != baseline_score_cap,
        "final_score": final_score,
        "formal_tier": tier,
        "score_delta": final_score - baseline_final,
        "raw_score_delta": raw_score - baseline_raw,
        "formal_score_changed": final_score != baseline_final,
        "notes": notes,
    }
    return simulation


def _apply_preview_deltas(profile: dict[str, Any], deltas: dict[str, Any]) -> None:
    for section, values in deltas.items():
        if isinstance(values, dict) and isinstance(profile.get(section), dict):
            profile[section].update(values)
        else:
            profile[section] = values


def _simulate_weighted_raw_score(result: dict[str, Any], profile: dict[str, Any]) -> int:
    weights = profile["weights"]
    score = result["score"]
    penalty = _simulated_c1_penalty(score, profile)
    weighted = (
        score["stage_1_readme_intent"] * weights["stage_1_percent"] / 100
        + score["stage_2_repo_local_consistency"] * weights["stage_2r_percent"] / 100
        + score["stage_3_code_bio"] * weights["stage_3_percent"] / 100
        - penalty
    )
    return round(weighted)


def _simulated_c1_penalty(score: dict[str, Any], profile: dict[str, Any]) -> int:
    baseline_penalty = int(score.get("risk_penalty", 0) or 0)
    if baseline_penalty <= 0:
        return 0
    return int(profile["code_integrity_policy"]["C1_penalty"])


def _simulate_score_cap(result: dict[str, Any], profile: dict[str, Any]) -> int | None:
    classification = result["classification"]
    clinical_policy = profile["clinical_policy"]
    if classification.get("t0_hard_floor"):
        return int(clinical_policy["t0_hard_floor_cap"])
    if classification.get("ca_severity") != "none" and not classification.get("has_explicit_clinical_boundary"):
        return int(clinical_policy["ca_no_disclaimer_cap"])
    return None


def _baseline_score_cap(result: dict[str, Any]) -> int | None:
    classification = result["classification"]
    if classification.get("t0_hard_floor"):
        return 39
    return classification.get("score_cap")


def _tier_from_policy(score: int, tier_policy: dict[str, Any]) -> str:
    boundaries = tier_policy["tier_boundaries"]
    names = tier_policy["tier_names"]
    labels = {
        "T0": "Rejected",
        "T1": "Quarantine",
        "T2": "Caution",
        "T3": "Supervised",
        "T4": "Candidate",
    }
    if score < boundaries[0]:
        tier_key = names[0]
    elif score < boundaries[1]:
        tier_key = names[1]
    elif score < boundaries[2]:
        tier_key = names[2]
    elif score < boundaries[3]:
        tier_key = names[3]
    else:
        tier_key = names[4]
    return f"{tier_key} {labels.get(tier_key, tier_key)}"