""" Risk Classifier. Evaluates domain nominations against global risk escalation rules to produce a final RiskAssessment with R0/R1/R2/R3 classification. Evaluation pipeline: 1. Check hard-escalate domains (suicidal_ideation → always R3) 2. For each nominated domain, evaluate risk_escalation_rules 3. Apply domain suppression rules (R1 wound_concern suppresses R2 wound_infection) 4. Select highest risk class across all active nominations 5. Determine recommended action (proceed / clarify / escalate / handoff) Safety invariants: - Hard-escalate domains CANNOT be suppressed - R3 can never be downgraded by suppression - When no rules match, default to the domain's target_risk_class - When no target_risk_class exists, default to R1 (fail-open) - Context-aware rules require explicit context; they fail closed (don't fire) when context is unavailable """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional, Set, Tuple from decision.engine.config_loader import DecisionConfigLoader from decision.engine.models import ( ConfidenceTier, DomainNomination, RiskAssessment, RiskClass, RiskRuleMatch, SuppressionResult, TurnOutcome, ) logger = logging.getLogger("decision.risk_classifier") class RiskClassifier: """ Classifies risk from domain nominations using global rules. Usage: classifier = RiskClassifier(config) assessment = classifier.assess( nominations=nominations, slot_values={"shortness_of_breath": "yes"}, patient_context=None, # Phase 3: FHIR data ) """ def __init__(self, config: DecisionConfigLoader): self._config = config self._hard_escalate: Set[str] = config.hard_escalate_domains self._safety_precedence: List[str] = config.safety_precedence self._risk_rules: List[Dict[str, Any]] = config.risk_escalation_rules self._suppression_rules: List[Dict[str, Any]] = config.domain_suppression_rules def assess( self, nominations: List[DomainNomination], slot_values: Optional[Dict[str, str]] = None, patient_context: Optional[Dict[str, Any]] = None, ml_label: Optional[str] = None, ml_confidence: Optional[float] = None, ) -> RiskAssessment: """ Produce a complete risk assessment from domain nominations. Args: nominations: Ranked domain nominations from DomainNominator slot_values: Currently filled slots (from flow execution) patient_context: FHIR-derived patient context (Phase 3) ml_label: Original DriveHealthBERT label ml_confidence: Original DriveHealthBERT confidence Returns: RiskAssessment with final risk class, matched rules, etc. """ if not nominations: return self._empty_assessment(ml_label, ml_confidence) slot_values = slot_values or {} active_nominations = [n for n in nominations if not n.is_negated] # Step 1: Check hard-escalate domains hard_escalate = False hard_domain = None for nom in active_nominations: if nom.domain in self._hard_escalate: hard_escalate = True hard_domain = nom.domain logger.warning( "HARD ESCALATE triggered: domain=%s", nom.domain ) break # Step 2: Evaluate risk rules for each nomination all_rule_matches: List[RiskRuleMatch] = [] for nom in active_nominations: matches = self._evaluate_rules_for_domain( nom.domain, slot_values, patient_context ) all_rule_matches.extend(matches) # Step 3: Apply domain suppression suppressions: List[SuppressionResult] = [] suppressed_domains: Set[str] = set() if not hard_escalate: suppressions, suppressed_domains = self._evaluate_suppressions( active_nominations ) # Step 4: Determine highest risk class risk_class = self._determine_risk_class( nominations=active_nominations, rule_matches=all_rule_matches, suppressed_domains=suppressed_domains, hard_escalate=hard_escalate, hard_domain=hard_domain, ) # Step 5: Determine primary domain primary_domain = self._select_primary_domain( active_nominations, suppressed_domains, hard_domain ) # Step 6: Determine recommended action recommended_action = self._determine_action(risk_class) # Step 7: Determine recommended flow from primary domain recommended_flow = None for nom in active_nominations: if nom.domain == primary_domain and nom.recommended_flow: recommended_flow = nom.recommended_flow break # Safety override flag safety_override = any( nom.confidence_tier >= ConfidenceTier.HIGH and nom.domain in set(self._safety_precedence) and nom.domain not in suppressed_domains for nom in active_nominations ) return RiskAssessment( risk_class=risk_class, primary_domain=primary_domain, domain_nominations=tuple(nominations), matched_rules=tuple(all_rule_matches), suppressions=tuple(suppressions), hard_escalate=hard_escalate, safety_override=safety_override, ml_label=ml_label, ml_confidence=ml_confidence, recommended_flow=recommended_flow, recommended_action=recommended_action, ) # ------------------------------------------------------------------ # Internal evaluation # ------------------------------------------------------------------ def _evaluate_rules_for_domain( self, domain: str, slot_values: Dict[str, str], patient_context: Optional[Dict[str, Any]], ) -> List[RiskRuleMatch]: """Evaluate all global risk rules for a specific domain.""" matches = [] for rule in self._risk_rules: rule_id = rule.get("id", "unknown") conditions = rule.get("if", {}) result = rule.get("then", {}) # Check domain match rule_domain = conditions.get("domain") if rule_domain != domain: continue # Check slot conditions slots_present = conditions.get("slots_present", []) slot_value_conditions = conditions.get("slot_values", {}) context_conditions = conditions.get("context") # All required slots must be present slots_ok = all( slot_name in slot_values for slot_name in slots_present ) # All slot value conditions must match values_ok = all( slot_values.get(k) == v for k, v in slot_value_conditions.items() ) # Context conditions (Phase 3) context_ok = True if context_conditions: if patient_context is None: # Fail closed: context required but not available context_ok = False else: context_ok = self._evaluate_context_conditions( context_conditions, patient_context ) if slots_ok and values_ok and context_ok: risk_str = result.get("risk_class", "R1") try: risk = RiskClass(risk_str) except ValueError: risk = RiskClass.R1 matches.append( RiskRuleMatch( rule_id=rule_id, domain=domain, risk_class=risk, conditions_met=slot_value_conditions, context_conditions=context_conditions, ) ) return matches def _evaluate_context_conditions( self, conditions: Dict[str, Any], patient_context: Dict[str, Any], ) -> bool: """ Evaluate FHIR context conditions against patient context. Phase 3: Full implementation. Currently supports: - patient_has_condition: list of ICD-10 patterns - medication_count_above: int threshold """ # patient_has_condition: check ICD-10 codes required_conditions = conditions.get("patient_has_condition", []) if required_conditions: patient_icd_codes = patient_context.get("active_conditions", []) if not self._match_icd_patterns(required_conditions, patient_icd_codes): return False # medication_count_above: check polypharmacy med_threshold = conditions.get("medication_count_above") if med_threshold is not None: med_count = patient_context.get("active_medication_count", 0) if med_count <= med_threshold: return False return True @staticmethod def _match_icd_patterns( patterns: List[str], patient_codes: List[str] ) -> bool: """Check if any patient ICD-10 code matches any required pattern.""" import re for pattern in patterns: # Convert ICD-10 wildcard to regex (e.g., "I50.*" → "I50\..*") regex = pattern.replace(".", r"\.").replace("*", ".*") for code in patient_codes: if re.match(regex, code, re.IGNORECASE): return True return False def _evaluate_suppressions( self, nominations: List[DomainNomination] ) -> Tuple[List[SuppressionResult], Set[str]]: """ Evaluate domain suppression rules. A lower-acuity domain firing at high confidence can suppress a higher-acuity domain at low/medium confidence to reduce false escalations. """ suppressions: List[SuppressionResult] = [] suppressed: Set[str] = set() active_domains = {n.domain: n for n in nominations} for rule in self._suppression_rules: suppressor_name = rule.get("suppressor") suppressed_list = rule.get("suppressed_domains", []) condition = rule.get("condition", {}) reason = rule.get("reason", "") suppressor = active_domains.get(suppressor_name) if not suppressor: continue # Check suppressor minimum confidence min_conf_str = condition.get("suppressor_min_confidence", "medium") min_conf = self._str_to_confidence(min_conf_str) if suppressor.confidence_tier < min_conf: continue # Check suppressed domains max_conf_str = condition.get("suppressed_max_confidence", "medium") max_conf = self._str_to_confidence(max_conf_str) for target_name in suppressed_list: target = active_domains.get(target_name) if not target: continue # SAFETY: Never suppress hard-escalate domains if target_name in self._hard_escalate: continue # Only suppress if target is at or below max confidence if target.confidence_tier <= max_conf: suppressed.add(target_name) suppressions.append( SuppressionResult( suppressed=True, suppressor_domain=suppressor_name, rule_reason=reason, ) ) logger.info( "Domain suppression: %s suppresses %s (reason: %s)", suppressor_name, target_name, reason, ) return suppressions, suppressed def _determine_risk_class( self, nominations: List[DomainNomination], rule_matches: List[RiskRuleMatch], suppressed_domains: Set[str], hard_escalate: bool, hard_domain: Optional[str], ) -> RiskClass: """Determine the highest applicable risk class.""" # Hard escalate always wins if hard_escalate: return RiskClass.R3 # Collect all risk classes from rule matches (excluding suppressed) risk_candidates: List[RiskClass] = [] for rm in rule_matches: if rm.domain not in suppressed_domains: risk_candidates.append(rm.risk_class) # ALSO consider target_risk_class from nominations (even when rules # matched for other domains). Previously this was a fallback that # only ran when zero rules matched, which let a benign R1 rule on # domain A mask a critical R3 target on domain B. for nom in nominations: if nom.domain in suppressed_domains: continue if nom.is_negated: continue if nom.target_risk_class: # For high-confidence matches on safety domains, use target risk if nom.confidence_tier >= ConfidenceTier.HIGH: risk_candidates.append(nom.target_risk_class) elif nom.confidence_tier >= ConfidenceTier.MEDIUM: # Medium confidence: one tier below target, minimum R1 downgraded = self._downgrade_risk(nom.target_risk_class) risk_candidates.append(downgraded) else: # Low confidence: two tiers below or R1 risk_candidates.append(RiskClass.R1) if risk_candidates: return max(risk_candidates) # Absolute fallback: if we have any non-negated nomination, R1 if any(not n.is_negated for n in nominations): return RiskClass.R1 return RiskClass.R0 def _select_primary_domain( self, nominations: List[DomainNomination], suppressed_domains: Set[str], hard_domain: Optional[str], ) -> Optional[str]: """Select the primary domain from active nominations.""" if hard_domain: return hard_domain # Use safety precedence order for tie-breaking precedence_set = set(self._safety_precedence) for nom in nominations: if nom.domain in suppressed_domains: continue if nom.is_negated: continue return nom.domain # Already sorted by confidence + priority # All negated or suppressed return nominations[0].domain if nominations else None @staticmethod def _determine_action(risk_class: RiskClass) -> TurnOutcome: """Map risk class to recommended turn outcome.""" if risk_class == RiskClass.R3: return TurnOutcome.ESCALATE elif risk_class == RiskClass.R2: return TurnOutcome.HANDOFF elif risk_class == RiskClass.R1: return TurnOutcome.PROCEED return TurnOutcome.PROCEED @staticmethod def _downgrade_risk(risk: RiskClass) -> RiskClass: """Downgrade risk by one tier, minimum R1.""" if risk == RiskClass.R3: return RiskClass.R2 elif risk == RiskClass.R2: return RiskClass.R1 return RiskClass.R1 @staticmethod def _str_to_confidence(s: str) -> ConfidenceTier: try: return ConfidenceTier(s) except ValueError: return ConfidenceTier.MEDIUM def _empty_assessment( self, ml_label: Optional[str] = None, ml_confidence: Optional[float] = None, ) -> RiskAssessment: """Return a default R0 assessment when no nominations exist.""" return RiskAssessment( risk_class=RiskClass.R0, primary_domain=None, domain_nominations=(), matched_rules=(), suppressions=(), hard_escalate=False, safety_override=False, ml_label=ml_label, ml_confidence=ml_confidence, recommended_flow=None, recommended_action=TurnOutcome.PROCEED, )