decision/engine/trigger_engine.py

"""
Taxonomy Trigger Engine.

Matches patient utterances against all 65 clinical domain trigger definitions
using phrase matching, regex patterns, partial substring matching, and
negation handling.

Safety-critical: This is the last line of defense for catching clinical
emergencies that the ML model may miss. False negatives here can be
life-threatening.

Design principles:
  - Fail open: If in doubt, nominate the domain (better to over-escalate)
  - Exhaustive matching: Check ALL domains, not just the first match
  - Negation requires explicit evidence: Only suppress if negation pattern
    clearly matches
  - Pre-compiled regex: All patterns compiled at config load time
  - Case-insensitive matching throughout
"""

from __future__ import annotations

import logging
import re
from typing import Any, Dict, List, Optional, Set, Tuple

from decision.engine.config_loader import DecisionConfigLoader
from decision.engine.models import (
    ConfidenceTier,
    MatchType,
    NegationAction,
    NegationResult,
    TriggerMatch,
)

logger = logging.getLogger("decision.trigger_engine")


class TaxonomyTriggerEngine:
    """
    Matches text against taxonomy trigger definitions.

    Usage:
        engine = TaxonomyTriggerEngine(config)
        matches = engine.match_all(text)
        # Returns: Dict[str, DomainMatchResult] keyed by domain name
    """

    def __init__(self, config: DecisionConfigLoader):
        self._config = config
        self._domains: Dict[str, Dict[str, Any]] = config.taxonomy_triggers
        logger.info(
            "TaxonomyTriggerEngine initialized with %d domains", len(self._domains)
        )

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def match_all(self, text: str) -> Dict[str, DomainMatchResult]:
        """
        Match text against ALL taxonomy domains.

        Returns a dict of domain_name -> DomainMatchResult for every domain
        that has at least one trigger match (before negation).

        Negation is evaluated but does NOT remove the domain from results.
        The caller decides what to do based on negation_result.
        """
        text_lower = text.lower().strip()
        if not text_lower:
            return {}

        results: Dict[str, DomainMatchResult] = {}

        for domain_name, triggers in self._domains.items():
            result = self._match_domain(domain_name, triggers, text, text_lower)
            if result and result.highest_confidence != ConfidenceTier.NONE:
                results[domain_name] = result

        return results

    def match_domain(self, domain_name: str, text: str) -> Optional[DomainMatchResult]:
        """Match text against a single domain's triggers."""
        triggers = self._domains.get(domain_name)
        if not triggers:
            return None
        text_lower = text.lower().strip()
        return self._match_domain(domain_name, triggers, text, text_lower)

    def check_negation(
        self, domain_name: str, text: str
    ) -> NegationResult:
        """Check if text matches negation patterns for a domain."""
        triggers = self._domains.get(domain_name, {})
        text_lower = text.lower().strip()
        return self._evaluate_negation(domain_name, triggers, text_lower)

    # ------------------------------------------------------------------
    # Internal matching
    # ------------------------------------------------------------------

    def _match_domain(
        self,
        domain_name: str,
        triggers: Dict[str, Any],
        text: str,
        text_lower: str,
    ) -> Optional[DomainMatchResult]:
        """Match text against a single domain's trigger definition."""
        lexical = triggers.get("lexical_signals", {})
        priority = triggers.get("priority", 0)

        all_matches: List[TriggerMatch] = []

        # Check each confidence tier (high → medium → low)
        for tier_name, tier_enum in [
            ("high", ConfidenceTier.HIGH),
            ("medium", ConfidenceTier.MEDIUM),
            ("low", ConfidenceTier.LOW),
        ]:
            tier = lexical.get(tier_name, {})
            tier_matches = self._match_tier(
                domain_name, tier, tier_enum, text, text_lower, priority
            )
            all_matches.extend(tier_matches)

        if not all_matches:
            return None

        # Determine highest confidence from matches
        highest = ConfidenceTier.NONE
        for m in all_matches:
            if m.confidence_tier > highest:
                highest = m.confidence_tier

        # Evaluate negation
        negation = self._evaluate_negation(domain_name, triggers, text_lower)

        # Apply negation to adjust confidence
        effective_confidence = highest
        if negation.is_negated:
            if negation.action == NegationAction.SUPPRESS:
                effective_confidence = ConfidenceTier.NONE
            elif negation.action == NegationAction.DOWNGRADE_CONFIDENCE:
                effective_confidence = self._downgrade_tier(highest)

        return DomainMatchResult(
            domain=domain_name,
            priority=priority,
            matches=tuple(all_matches),
            highest_confidence=highest,
            effective_confidence=effective_confidence,
            negation_result=negation,
        )

    def _match_tier(
        self,
        domain_name: str,
        tier: Dict[str, Any],
        tier_enum: ConfidenceTier,
        text: str,
        text_lower: str,
        priority: int,
    ) -> List[TriggerMatch]:
        """Match text against a single confidence tier."""
        matches: List[TriggerMatch] = []

        # 1. Phrase matching (exact substring, case-insensitive)
        for phrase in tier.get("phrases", []):
            phrase_lower = phrase.lower()
            idx = text_lower.find(phrase_lower)
            if idx >= 0:
                matched_span = text[idx : idx + len(phrase)]
                matches.append(
                    TriggerMatch(
                        domain=domain_name,
                        confidence_tier=tier_enum,
                        match_type=MatchType.PHRASE,
                        matched_text=phrase,
                        matched_span=matched_span,
                        priority=priority,
                    )
                )

        # 2. Regex matching (pre-compiled)
        for compiled_re in tier.get("_compiled_regex", []):
            m = compiled_re.search(text)
            if m:
                matches.append(
                    TriggerMatch(
                        domain=domain_name,
                        confidence_tier=tier_enum,
                        match_type=MatchType.REGEX,
                        matched_text=compiled_re.pattern,
                        matched_span=m.group(0),
                        priority=priority,
                    )
                )

        # 3. Partial matching (substring, case-insensitive)
        for partial in tier.get("partials", []):
            partial_lower = partial.lower()
            idx = text_lower.find(partial_lower)
            if idx >= 0:
                matched_span = text[idx : idx + len(partial)]
                matches.append(
                    TriggerMatch(
                        domain=domain_name,
                        confidence_tier=tier_enum,
                        match_type=MatchType.PARTIAL,
                        matched_text=partial,
                        matched_span=matched_span,
                        priority=priority,
                    )
                )

        return matches

    def _evaluate_negation(
        self,
        domain_name: str,
        triggers: Dict[str, Any],
        text_lower: str,
    ) -> NegationResult:
        """
        Evaluate negation patterns for a domain.

        SAFETY DESIGN: Negation requires an EXPLICIT match against a known
        negation pattern. We do NOT use generic "no/not" detection because
        that risks suppressing true emergencies.
        """
        negation_config = triggers.get("negation_handling", {})
        compiled_patterns: List[str] = negation_config.get("_compiled_patterns", [])
        action_str = negation_config.get("action", "downgrade_confidence")

        try:
            action = NegationAction(action_str)
        except ValueError:
            action = NegationAction.DOWNGRADE_CONFIDENCE

        for pattern in compiled_patterns:
            if pattern in text_lower:
                return NegationResult(
                    is_negated=True,
                    action=action,
                    matched_pattern=pattern,
                )

        return NegationResult(
            is_negated=False,
            action=action,
        )

    @staticmethod
    def _downgrade_tier(tier: ConfidenceTier) -> ConfidenceTier:
        """Downgrade confidence by one level."""
        if tier == ConfidenceTier.HIGH:
            return ConfidenceTier.MEDIUM
        elif tier == ConfidenceTier.MEDIUM:
            return ConfidenceTier.LOW
        elif tier == ConfidenceTier.LOW:
            return ConfidenceTier.NONE
        return ConfidenceTier.NONE


# ---------------------------------------------------------------------------
# Domain Match Result
# ---------------------------------------------------------------------------

class DomainMatchResult:
    """
    Result of matching a single domain against patient text.

    Contains all trigger matches, the highest raw confidence,
    effective confidence (after negation), and negation details.
    """

    __slots__ = (
        "domain",
        "priority",
        "matches",
        "highest_confidence",
        "effective_confidence",
        "negation_result",
    )

    def __init__(
        self,
        domain: str,
        priority: int,
        matches: Tuple[TriggerMatch, ...],
        highest_confidence: ConfidenceTier,
        effective_confidence: ConfidenceTier,
        negation_result: NegationResult,
    ):
        self.domain = domain
        self.priority = priority
        self.matches = matches
        self.highest_confidence = highest_confidence
        self.effective_confidence = effective_confidence
        self.negation_result = negation_result

    @property
    def is_negated(self) -> bool:
        return self.negation_result.is_negated

    @property
    def is_suppressed(self) -> bool:
        return self.effective_confidence == ConfidenceTier.NONE

    @property
    def match_count(self) -> int:
        return len(self.matches)

    def __repr__(self) -> str:
        neg = " [NEGATED]" if self.is_negated else ""
        return (
            f"DomainMatchResult({self.domain}, "
            f"confidence={self.effective_confidence.value}, "
            f"priority={self.priority}, "
            f"matches={self.match_count}{neg})"
        )