server/my_env_environment.py

# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

"""
MedCodeRL Environment Implementation.

Medical Coding & Billing Compliance environment where agents must:
1. Assign correct ICD-10 diagnosis codes
2. Assign correct CPT procedure codes
3. Make billing compliance decisions (approve/reject/review)
4. Provide clinical reasoning
5. Identify compliance risks

Contains: environment logic, deterministic grader, shaped rewards, action validation.
"""

import json
import math
import os
import re
from typing import Any, Dict, List, Optional, Set, Tuple
from uuid import uuid4

from openenv.core.env_server.interfaces import Environment
from openenv.core.env_server.types import State

try:
    from ..models import MedAction, MedObservation
except ImportError:
    from models import MedAction, MedObservation


# ──────────────────────────────────────────────
#  Task loader
# ──────────────────────────────────────────────

TASKS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tasks")


def _load_task_cases(difficulty: str) -> List[Dict[str, Any]]:
    """Load clinical cases for a given difficulty level."""
    filepath = os.path.join(TASKS_DIR, f"{difficulty}.json")
    if not os.path.exists(filepath):
        return []
    with open(filepath, "r", encoding="utf-8") as f:
        data = json.load(f)
    return data.get("cases", [])


# ──────────────────────────────────────────────
#  Action validation
# ──────────────────────────────────────────────

ICD10_PATTERN = re.compile(r"^[A-Z]\d{2}(\.\d{1,4})?$", re.IGNORECASE)
CPT_PATTERN = re.compile(r"^\d{5}$")
HCPCS_PATTERN = re.compile(r"^[A-Z]\d{4}$", re.IGNORECASE)
SCORE_EPSILON = 0.01


def _to_open_interval_score(value: float) -> float:
    """Map a score to the strict open interval (0, 1).

    Guarantees the returned value satisfies  0 < value < 1.
    """
    try:
        score = float(value)
    except (TypeError, ValueError):
        score = 0.0

    if not math.isfinite(score):
        score = 0.0

    # Clamp into the safe open interval (SCORE_EPSILON, 1 - SCORE_EPSILON)
    score = max(SCORE_EPSILON, min(1.0 - SCORE_EPSILON, score))
    return score


def _rounded_open_interval_score(value: float, ndigits: int = 4) -> float:
    """Round score while preserving strict open interval bounds."""
    clamped = _to_open_interval_score(value)
    rounded = round(clamped, ndigits)
    # Re-clamp after rounding to guarantee strict (0, 1)
    return _to_open_interval_score(rounded)


def _rounded_component_score(value: float, ndigits: int = 4) -> float:
    """Normalize component scores to strict open interval for validator safety."""
    return _rounded_open_interval_score(value, ndigits)


def _validate_action(action_dict: dict) -> Tuple[bool, List[str]]:
    """Validate an action dict. Returns (is_valid, error_list)."""
    errors: List[str] = []

    diag = action_dict.get("diagnosis_codes", [])
    if not isinstance(diag, list) or len(diag) == 0:
        errors.append("At least one diagnosis code is required")
    elif len(diag) > 5:
        errors.append("Maximum 5 diagnosis codes allowed")
    else:
        for code in diag:
            if not ICD10_PATTERN.match(str(code).strip()):
                errors.append(f"Invalid ICD-10 format: {code}")

    proc = action_dict.get("procedure_codes", [])
    if isinstance(proc, list):
        for code in proc:
            c = str(code).strip()
            if not CPT_PATTERN.match(c) and not HCPCS_PATTERN.match(c):
                errors.append(f"Invalid CPT/HCPCS format: {code}")

    decision = action_dict.get("decision", "")
    if str(decision).lower() not in ("approve", "reject", "review"):
        errors.append(f"Invalid decision: {decision}")

    confidence = action_dict.get("confidence", -1)
    try:
        conf_val = float(confidence)
        if conf_val < 0.0 or conf_val > 1.0:
            errors.append("confidence must be 0.0-1.0")
    except (TypeError, ValueError):
        errors.append("confidence must be a number")

    reasoning = action_dict.get("reasoning", "")
    if not isinstance(reasoning, str) or len(reasoning) < 15:
        errors.append("reasoning must be at least 15 characters")

    return len(errors) == 0, errors


# ──────────────────────────────────────────────
#  Deterministic grader
# ──────────────────────────────────────────────

def _set_similarity(predicted: List[str], ground_truth: List[str]) -> float:
    """Jaccard similarity between two code sets."""
    if not ground_truth and not predicted:
        return 1.0
    pred_set: Set[str] = set(c.strip().upper() for c in predicted if c)
    gt_set: Set[str] = set(c.strip().upper() for c in ground_truth if c)
    if not gt_set:
        return 1.0 if not pred_set else 0.0
    if not pred_set:
        return 0.0
    intersection = pred_set & gt_set
    union = pred_set | gt_set
    return len(intersection) / len(union) if union else 1.0


def _partial_code_match(predicted: List[str], ground_truth: List[str]) -> float:
    """Partial-credit matching for medical codes (prefix similarity)."""
    if not ground_truth:
        return 1.0 if not predicted else 0.0
    if not predicted:
        return 0.0
    pred_list = [c.strip().upper() for c in predicted if c]
    gt_list = [c.strip().upper() for c in ground_truth if c]
    total = 0.0
    for gt_code in gt_list:
        best = 0.0
        for pred_code in pred_list:
            if pred_code == gt_code:
                best = 1.0
                break
            gt_base = gt_code.split(".")[0]
            pred_base = pred_code.split(".")[0]
            if gt_base == pred_base:
                best = max(best, 0.5)
            elif len(gt_base) >= 3 and gt_base[:3] == pred_base[:3]:
                best = max(best, 0.25)
        total += best
    return total / len(gt_list)


def _grade(action_dict: dict, ground_truth: dict) -> Dict[str, float]:
    """
    Deterministic grading — 6 weighted components → score in [0.0, 1.0].

    Weights: diagnosis 35%, procedure 20%, decision 25%, reasoning 10%,
    risk flags 5%, confidence calibration 5%.
    """
    pred_diag = action_dict.get("diagnosis_codes", [])
    gt_diag = ground_truth.get("diagnosis_codes", [])
    pred_proc = action_dict.get("procedure_codes", [])
    gt_proc = ground_truth.get("procedure_codes", [])
    pred_dec = str(action_dict.get("decision", "")).lower()
    gt_dec = str(ground_truth.get("decision", "")).lower()
    pred_reasoning = str(action_dict.get("reasoning", ""))
    pred_risk = action_dict.get("risk_flags", [])
    gt_risk = ground_truth.get("risk_flags", [])
    pred_conf = float(action_dict.get("confidence", 0.5))

    # 1. Diagnosis codes (35%)
    diag_exact = _set_similarity(pred_diag, gt_diag)
    diag_partial = _partial_code_match(pred_diag, gt_diag)
    diag_score = 0.6 * diag_exact + 0.4 * diag_partial

    # 2. Procedure codes (20%)
    proc_exact = _set_similarity(pred_proc, gt_proc)
    proc_partial = _partial_code_match(pred_proc, gt_proc)
    proc_score = 0.6 * proc_exact + 0.4 * proc_partial

    # 3. Decision (25%)
    if pred_dec == gt_dec:
        dec_score = 1.0
    elif pred_dec == "review" and gt_dec in ("approve", "reject"):
        dec_score = 0.3
    elif pred_dec in ("approve", "reject") and gt_dec == "review":
        dec_score = 0.2
    else:
        dec_score = 0.0

    # 4. Reasoning quality (10%)
    reasoning_lower = pred_reasoning.lower()
    r_score = 0.0
    if len(pred_reasoning) >= 20:
        r_score += 0.3
    if len(pred_reasoning) >= 50:
        r_score += 0.2
    med_terms = [
        "icd", "cpt", "diagnosis", "procedure", "coding", "compliance",
        "medical", "clinical", "treatment", "patient", "billing",
        "authorization", "insurance", "modifier", "documentation",
        "justified", "appropriate", "medically necessary", "guideline",
    ]
    r_score += min(0.5, sum(1 for t in med_terms if t in reasoning_lower) * 0.1)
    r_score = min(1.0, r_score)

    # 5. Risk flags (5%)
    risk_score = _set_similarity(pred_risk, gt_risk)

    # 6. Confidence calibration (5%)
    correctness = diag_score * 0.5 + proc_score * 0.3 + dec_score * 0.2
    conf_score = max(0.0, 1.0 - abs(pred_conf - correctness) * 2.0)

    total = (
        diag_score * 0.35
        + proc_score * 0.20
        + dec_score * 0.25
        + r_score * 0.10
        + risk_score * 0.05
        + conf_score * 0.05
    )

    return {
        "score": _rounded_open_interval_score(total, 4),
        "diagnosis_accuracy": _rounded_component_score(diag_score, 4),
        "procedure_accuracy": _rounded_component_score(proc_score, 4),
        "decision_accuracy": _rounded_component_score(dec_score, 4),
        "reasoning_quality": _rounded_component_score(r_score, 4),
        "risk_identification": _rounded_component_score(risk_score, 4),
        "confidence_calibration": _rounded_component_score(conf_score, 4),
    }


# ──────────────────────────────────────────────
#  Shaped reward engine
# ──────────────────────────────────────────────

def _compute_reward(action_dict: dict, ground_truth: dict, difficulty: str = "easy") -> Dict:
    """
    Shaped reward = base_grade + bonuses − penalties.

    Penalties: upcoding, undercoding, wrong denial/approval, unnecessary procedure,
    missing primary code, low confidence.
    Bonuses: perfect diagnosis, good reasoning, all risk flags.
    """
    grade_result = _grade(action_dict, ground_truth)
    base = grade_result["score"]

    pred_diag = set(c.strip().upper() for c in action_dict.get("diagnosis_codes", []) if c)
    gt_diag = set(c.strip().upper() for c in ground_truth.get("diagnosis_codes", []) if c)
    gt_diag_list = [c.strip().upper() for c in ground_truth.get("diagnosis_codes", []) if c]
    pred_proc = set(c.strip().upper() for c in action_dict.get("procedure_codes", []) if c)
    gt_proc = set(c.strip().upper() for c in ground_truth.get("procedure_codes", []) if c)
    pred_dec = str(action_dict.get("decision", "")).lower()
    gt_dec = str(ground_truth.get("decision", "")).lower()
    pred_conf = float(action_dict.get("confidence", 0.5))

    penalties: Dict[str, float] = {}
    bonuses: Dict[str, float] = {}

    # Penalties
    if len(pred_proc) > len(gt_proc) + 1:
        penalties["upcoding"] = -0.15
    if gt_diag and len(pred_diag & gt_diag) < len(gt_diag) * 0.5:
        penalties["undercoding"] = -0.10
    if pred_dec == "reject" and gt_dec == "approve":
        penalties["wrong_denial"] = -0.20
    if pred_dec == "approve" and gt_dec == "reject":
        penalties["wrong_approval"] = -0.25
    if pred_proc - gt_proc:
        penalties["unnecessary_procedure"] = -0.10
    if gt_diag_list and gt_diag_list[0] not in pred_diag:
        penalties["missing_primary_code"] = -0.15
    if pred_conf < 0.2:
        penalties["low_confidence"] = -0.05

    # Bonuses
    if grade_result["diagnosis_accuracy"] >= 0.99:
        bonuses["perfect_diagnosis"] = 0.05
    if grade_result["reasoning_quality"] >= 0.8:
        bonuses["good_reasoning"] = 0.03
    if grade_result["risk_identification"] >= 0.99:
        bonuses["all_risk_flags"] = 0.05

    diff_mult = {"easy": 0.8, "medium": 1.0, "hard": 1.2}.get(difficulty, 1.0)
    total_penalty = sum(penalties.values()) * diff_mult
    total_bonus = sum(bonuses.values())
    final = _to_open_interval_score(base + total_penalty + total_bonus)

    feedback_parts = []
    if penalties:
        feedback_parts.append(f"Penalties: {', '.join(penalties.keys())}")
    if bonuses:
        feedback_parts.append(f"Bonuses: {', '.join(bonuses.keys())}")
    if not penalties and not bonuses:
        feedback_parts.append("Clean submission.")

    return {
        "score": _rounded_open_interval_score(final, 4),
        "breakdown": {
            "base_grade": _rounded_open_interval_score(base, 4),
            "grade_components": grade_result,
        },
        "feedback": " | ".join(feedback_parts),
    }


# ──────────────────────────────────────────────
#  Core Environment
# ──────────────────────────────────────────────

class MyEnvironment(Environment):
    """
    MedCodeRL — Medical Coding & Billing Compliance Environment.

    90 realistic clinical cases (30 easy, 30 medium, 30 hard) covering:
    - Straightforward coding (easy)
    - Multi-diagnosis with comorbidities (medium)
    - Compliance dilemmas: upcoding, unbundling, fraud, ethical edge cases (hard)

    OpenEnv-compliant: reset() / step() / state property.
    """

    SUPPORTS_CONCURRENT_SESSIONS: bool = True

    def __init__(self):
        """Initialize the MedCodeRL environment."""
        self._state = State(episode_id=str(uuid4()), step_count=0)
        self._reset_count = 0

        # Load all task cases
        self._task_cases: Dict[str, List[Dict]] = {}
        self._case_index: Dict[str, int] = {"easy": 0, "medium": 0, "hard": 0}
        for diff in ("easy", "medium", "hard"):
            self._task_cases[diff] = _load_task_cases(diff)

        self._current_case: Optional[Dict] = None
        self._current_difficulty: str = "easy"
        self._done: bool = True
        self._action_history: List[dict] = []

    def _pick_case(self, task_id: Optional[str] = None) -> Dict:
        """Select a case by difficulty or specific case_id."""
        if task_id in ("easy", "medium", "hard"):
            difficulty = task_id
        elif task_id:
            # Search for specific case_id
            for diff in ("easy", "medium", "hard"):
                for case in self._task_cases.get(diff, []):
                    if case["id"] == task_id:
                        self._current_difficulty = diff
                        return case
            raise ValueError(f"Case not found: {task_id}")
        else:
            # Cycle through difficulties
            total = sum(self._case_index.values())
            difficulty = ["easy", "medium", "hard"][total % 3]

        cases = self._task_cases.get(difficulty, [])
        if not cases:
            raise ValueError(f"No cases for difficulty: {difficulty}")
        idx = self._case_index[difficulty] % len(cases)
        self._case_index[difficulty] = idx + 1
        self._current_difficulty = difficulty
        return cases[idx]

    def _build_observation(self, case: Dict, done: bool = False,
                           reward: Optional[float] = None,
                           reward_breakdown: Optional[Dict] = None,
                           feedback: str = "") -> MedObservation:
        """Build a MedObservation from a case dict."""
        inp = case.get("input", case)
        return MedObservation(
            case_id=case.get("id", ""),
            difficulty=case.get("difficulty", self._current_difficulty),
            clinical_note=inp.get("clinical_note", ""),
            symptoms=inp.get("symptoms", []),
            treatments=inp.get("treatments", []),
            insurance_type=inp.get("insurance_type", "Private"),
            prior_auth_required=inp.get("prior_auth_required", False),
            treatment_cost=inp.get("treatment_cost", "low"),
            patient_age=inp.get("patient_age", 0),
            patient_sex=inp.get("patient_sex", "M"),
            provider_specialty=inp.get("provider_specialty", ""),
            visit_type=inp.get("visit_type", "outpatient"),
            comorbidities=inp.get("comorbidities", []),
            lab_results=inp.get("lab_results"),
            medications=inp.get("medications", []),
            reward_breakdown=reward_breakdown,
            feedback=feedback,
            done=done,
            reward=reward,
            metadata={
                "step_count": self._state.step_count,
                "difficulty": self._current_difficulty,
            },
        )

    def reset(self, task_id: Optional[str] = None, **kwargs) -> MedObservation:
        """
        Reset the environment to a new episode.

        Args:
            task_id: 'easy', 'medium', 'hard', or a specific case_id.

        Returns:
            MedObservation with the clinical case to code.
        """
        # Accept task_id from kwargs if not provided directly
        if task_id is None:
            task_id = kwargs.get("task_id")

        self._state = State(episode_id=str(uuid4()), step_count=0)
        self._reset_count += 1
        self._done = False
        self._action_history = []

        self._current_case = self._pick_case(task_id)

        return self._build_observation(
            self._current_case,
            done=False,
            reward=_to_open_interval_score(0.0),
        )

    def step(self, action: MedAction) -> MedObservation:  # type: ignore[override]
        """
        Execute a step: grade the agent's medical coding action.

        Args:
            action: MedAction with diagnosis codes, procedure codes, decision, etc.

        Returns:
            MedObservation with reward and grading breakdown.
        """
        if self._done:
            return self._build_observation(
                self._current_case or {},
                done=True,
                reward=_to_open_interval_score(0.0),
                feedback="Episode already done. Call reset().",
            )

        self._state.step_count += 1

        # Convert action to dict
        action_dict = {
            "diagnosis_codes": action.diagnosis_codes,
            "procedure_codes": action.procedure_codes,
            "decision": action.decision,
            "confidence": action.confidence,
            "reasoning": action.reasoning,
            "modifier_codes": action.modifier_codes,
            "risk_flags": action.risk_flags,
        }

        # Validate
        is_valid, errors = _validate_action(action_dict)

        if not is_valid:
            self._action_history.append({"action": action_dict, "valid": False})
            if self._state.step_count >= 3:
                self._done = True
            return self._build_observation(
                self._current_case or {},
                done=self._done,
                reward=_to_open_interval_score(0.0),
                feedback=f"Invalid action: {'; '.join(errors)}",
            )

        # Grade against ground truth
        ground_truth = self._current_case.get("ground_truth", {})
        reward_result = _compute_reward(action_dict, ground_truth, self._current_difficulty)

        self._action_history.append({"action": action_dict, "valid": True})
        self._done = True  # single-step episode for valid actions

        score = _to_open_interval_score(reward_result["score"])

        return self._build_observation(
            self._current_case or {},
            done=True,
            reward=score,
            reward_breakdown=reward_result.get("breakdown"),
            feedback=reward_result.get("feedback", ""),
        )

    @property
    def state(self) -> State:
        """
        Get the current environment state.

        Returns:
            State with episode_id, step_count, and metadata
        """
        self._state.metadata = {
            "done": self._done,
            "difficulty": self._current_difficulty,
            "case_id": self._current_case.get("id", "") if self._current_case else "",
            "reset_count": self._reset_count,
        }
        return self._state

    def get_task_info(self) -> dict:
        """Return task metadata for discovery endpoints."""
        return {
            "tasks": {
                diff: {
                    "count": len(cases),
                    "description": {
                        "easy": "Straightforward cases with single diagnoses and direct ICD-10/CPT mapping.",
                        "medium": "Multi-diagnosis cases with comorbidities, insurance considerations, and partial ambiguity.",
                        "hard": "Complex compliance dilemmas: upcoding, unbundling, fraud detection, ethical edge cases.",
                    }.get(diff, ""),
                }
                for diff, cases in self._task_cases.items()
            },
            "total_cases": sum(len(c) for c in self._task_cases.values()),
        }