service/app.py

#!/usr/bin/env python3
"""
FastAPI Inference Service for DCE Intent BioClinicalBERT.

Low-latency inference using ONNX Runtime INT8 quantized model.
Scalability features: async inference, connection pooling, metrics, batching.

Usage:
    API_KEY=your-secret-key uvicorn service.app:app --host 0.0.0.0 --port 8000
    
    # Or specify a different port:
    API_KEY=your-secret-key uvicorn service.app:app --host 0.0.0.0 --port 8001
    
    # Production with multiple workers:
    API_KEY=your-secret-key gunicorn service.app:app -w 4 -k uvicorn.workers.UvicornWorker
"""

import asyncio
import hashlib
import json
import logging
import os
import re
import secrets
import signal
import sys
import threading
import time
import uuid
from collections import OrderedDict
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import bleach
import numpy as np
import onnxruntime as ort
from fastapi import Depends, FastAPI, HTTPException, Query, Request, Security, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, Field, field_validator
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
from transformers import AutoTokenizer

# Import HIPAA compliance utilities
try:
    from service.hipaa_config import (
        hipaa_audit,
        redact_phi,
        check_clinical_safety_keywords,
        get_security_headers,
        get_compliance_status,
        REQUIRE_TLS,
    )
    HIPAA_ENABLED = True
except ImportError:
    HIPAA_ENABLED = False

# Import Temporal Preprocessor
try:
    from service.temporal_preprocessor import preprocess_temporal, TemporalSignal
    TEMPORAL_PREPROCESSOR_AVAILABLE = True
except ImportError:
    TEMPORAL_PREPROCESSOR_AVAILABLE = False

# Import Taxonomy Scorer
try:
    from service.taxonomy_scorer import score_taxonomy, TaxonomyScoreResult
    TAXONOMY_SCORER_AVAILABLE = True
except ImportError:
    TAXONOMY_SCORER_AVAILABLE = False

# Import Decision Engine (Phase 1: Safety Trifecta)
try:
    from decision.engine.config_loader import DecisionConfigLoader
    from decision.engine.trigger_engine import TaxonomyTriggerEngine
    from decision.engine.domain_nomination import DomainNominator
    from decision.engine.risk_classifier import RiskClassifier
    from decision.engine.models import RiskClass, ConfidenceTier, RiskAssessment
    from decision.engine.flow_engine import FlowEngine, TurnResult
    from decision.engine.fhir_context import PatientContextBuilder
    from decision.engine.tenant_manager import TenantManager
    from decision.engine.journey_orchestrator import JourneyOrchestrator
    DECISION_ENGINE_AVAILABLE = True
except ImportError as _de_err:
    DECISION_ENGINE_AVAILABLE = False
    logger.warning("Decision engine not available: %s", _de_err)

# =============================================================================
# CONTEXT VARIABLES FOR THREAD-SAFE LOGGING
# =============================================================================

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
client_ip_var: ContextVar[str] = ContextVar("client_ip", default="-")


class RequestContextFilter(logging.Filter):
    """Add request context to log records using context variables (thread-safe)."""
    def filter(self, record):
        record.request_id = request_id_var.get()
        record.client_ip = client_ip_var.get()
        return True


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - [%(request_id)s] [%(client_ip)s] - %(message)s",
)
logger = logging.getLogger(__name__)
logger.addFilter(RequestContextFilter())

# Audit logger for security events
audit_logger = logging.getLogger("audit")
audit_logger.setLevel(logging.INFO)
audit_logger.addFilter(RequestContextFilter())

# =============================================================================
# SECURITY CONFIGURATION
# =============================================================================

# API Key Authentication
API_KEY = os.getenv("API_KEY", "")  # Required in production
API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
REQUIRE_AUTH = os.getenv("REQUIRE_AUTH", "false").lower() == "true"

# Rate Limiting Configuration
RATE_LIMIT = os.getenv("RATE_LIMIT", "100/minute")  # Requests per minute
# Use Redis for multi-worker deployments: redis://localhost:6379
# Falls back to memory for single-worker or development
RATE_LIMIT_STORAGE_URI = os.getenv("RATE_LIMIT_STORAGE", "memory://")
if "gunicorn" in os.getenv("SERVER_SOFTWARE", "") and RATE_LIMIT_STORAGE_URI == "memory://":
    logger.warning("Rate limiting with memory storage won't work across gunicorn workers. Set RATE_LIMIT_STORAGE=redis://...")

# CORS Configuration - comma-separated list of allowed origins
ALLOWED_ORIGINS = os.getenv("ALLOWED_ORIGINS", "http://localhost:3000,http://localhost:8080").split(",")
ALLOWED_ORIGINS = [origin.strip() for origin in ALLOWED_ORIGINS if origin.strip()]

# Path validation - allowed base directories for model loading
ALLOWED_MODEL_PATHS = [
    Path("artifacts").resolve(),
    Path(os.getenv("ALLOWED_MODEL_BASE", "artifacts")).resolve(),
]

# =============================================================================
# SCALABILITY CONFIGURATION
# =============================================================================

# Thread pool for CPU-bound inference (prevents blocking async event loop)
INFERENCE_WORKERS = int(os.getenv("INFERENCE_WORKERS", "4"))
inference_executor: Optional[ThreadPoolExecutor] = None

# ONNX Runtime session configuration
ONNX_INTRA_OP_THREADS = int(os.getenv("ONNX_INTRA_OP_THREADS", "4"))
ONNX_INTER_OP_THREADS = int(os.getenv("ONNX_INTER_OP_THREADS", "1"))
ONNX_GRAPH_OPT_LEVEL = os.getenv("ONNX_GRAPH_OPT_LEVEL", "all")  # all, basic, extended, disabled

# Request batching configuration (optional, for high-throughput scenarios)
ENABLE_BATCHING = os.getenv("ENABLE_BATCHING", "false").lower() == "true"
BATCH_SIZE = int(os.getenv("BATCH_SIZE", "8"))
BATCH_TIMEOUT_MS = int(os.getenv("BATCH_TIMEOUT_MS", "10"))

# Metrics configuration
ENABLE_METRICS = os.getenv("ENABLE_METRICS", "true").lower() == "true"

# Request timeout configuration (seconds)
INFERENCE_TIMEOUT_SECONDS = float(os.getenv("INFERENCE_TIMEOUT", "30.0"))

# Request body size limit (bytes) - prevents DoS via large payloads
MAX_REQUEST_BODY_SIZE = int(os.getenv("MAX_REQUEST_BODY_SIZE", str(1024 * 1024)))  # 1MB default

# Default decision policy constants (used when policy file not found)
DEFAULT_ESCALATION_LABEL = "ESCALATION"
DEFAULT_ESCALATION_THRESHOLD = 0.5
DEFAULT_MIN_CONFIDENCE = 0.3
DEFAULT_FALLBACK_LABEL = "OTHER"

# =============================================================================
# METRICS TRACKING
# =============================================================================

@dataclass
class ServiceMetrics:
    """Thread-safe metrics for monitoring service performance."""
    total_requests: int = 0
    successful_requests: int = 0
    failed_requests: int = 0
    total_inference_time_ms: float = 0.0
    total_tokenize_time_ms: float = 0.0
    escalations_triggered: int = 0
    start_time: float = field(default_factory=time.time)
    _lock: Any = field(default_factory=lambda: __import__('threading').Lock())
    
    def record_request(self, success: bool, inference_ms: float, tokenize_ms: float, escalated: bool):
        """Record metrics for a request (thread-safe with lock)."""
        with self._lock:
            self.total_requests += 1
            if success:
                self.successful_requests += 1
            else:
                self.failed_requests += 1
            self.total_inference_time_ms += inference_ms
            self.total_tokenize_time_ms += tokenize_ms
            if escalated:
                self.escalations_triggered += 1
    
    def get_stats(self) -> Dict[str, Any]:
        """Get current metrics as dictionary."""
        uptime = time.time() - self.start_time
        avg_inference = self.total_inference_time_ms / max(1, self.successful_requests)
        avg_tokenize = self.total_tokenize_time_ms / max(1, self.successful_requests)
        return {
            "uptime_seconds": round(uptime, 2),
            "total_requests": self.total_requests,
            "successful_requests": self.successful_requests,
            "failed_requests": self.failed_requests,
            "success_rate": round(self.successful_requests / max(1, self.total_requests), 4),
            "escalations_triggered": self.escalations_triggered,
            "avg_inference_ms": round(avg_inference, 3),
            "avg_tokenize_ms": round(avg_tokenize, 3),
            "requests_per_second": round(self.total_requests / max(1, uptime), 2),
        }


metrics = ServiceMetrics()

# =============================================================================
# MODEL CONFIGURATION
# =============================================================================

def validate_path(path: Path, allowed_bases: List[Path]) -> Path:
    """Validate that a path is within allowed directories (prevent traversal)."""
    resolved = path.resolve()
    for base in allowed_bases:
        try:
            resolved.relative_to(base)
            return resolved
        except ValueError:
            continue
    raise ValueError(f"Path {path} is not within allowed directories")

# Configuration from environment with path validation
_model_dir_raw = Path(os.getenv("MODEL_DIR", "artifacts/onnx_int8"))
_environment = os.getenv("ENVIRONMENT", "development").lower()
try:
    MODEL_DIR = validate_path(_model_dir_raw, ALLOWED_MODEL_PATHS)
except ValueError:
    if _environment == "production":
        raise RuntimeError(
            f"MODEL_DIR path validation failed for '{_model_dir_raw}'. "
            "In production, MODEL_DIR must be within allowed model paths."
        )
    # Fall back to default only in non-production
    MODEL_DIR = Path("artifacts/onnx_int8").resolve()
    logger.warning(f"MODEL_DIR path validation failed, using default: {MODEL_DIR}")

_policy_path_raw = Path(os.getenv("POLICY_PATH", "artifacts/config/decision_policy.json"))
try:
    POLICY_PATH = validate_path(_policy_path_raw, ALLOWED_MODEL_PATHS)
except ValueError:
    if _environment == "production":
        raise RuntimeError(
            f"POLICY_PATH path validation failed for '{_policy_path_raw}'. "
            "In production, POLICY_PATH must be within allowed model paths."
        )
    POLICY_PATH = Path("artifacts/config/decision_policy.json").resolve()
    logger.warning(f"POLICY_PATH path validation failed, using default: {POLICY_PATH}")

MAX_LENGTH = int(os.getenv("MAX_LENGTH", "64"))  # Match v4 BioClinicalBERT training config
MAX_INPUT_CHARS = int(os.getenv("MAX_INPUT_CHARS", "1024"))
MAX_COMBINED_CHARS = int(os.getenv("MAX_COMBINED_CHARS", "1500"))  # text + context limit

# =============================================================================
# RATE LIMITER SETUP
# =============================================================================

limiter = Limiter(key_func=get_remote_address)

# =============================================================================
# FASTAPI APPLICATION
# =============================================================================

# Conditionally disable docs in production
DISABLE_DOCS = os.getenv("DISABLE_DOCS", "false").lower() == "true"

app = FastAPI(
    title="Medical Intent Escalation",
    description="DCE version 1.0.0",
    version="1.0.0",
    docs_url=None if DISABLE_DOCS else "/docs",
    redoc_url=None if DISABLE_DOCS else "/redoc",
    openapi_url=None if DISABLE_DOCS else "/openapi.json",
)

# Root redirect to Swagger docs
@app.get("/", include_in_schema=False)
async def root():
    return RedirectResponse(url="/docs")

# Add rate limiter to app
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# CORS middleware with restricted origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["POST", "GET"],
    allow_headers=["X-API-Key", "Content-Type", "Authorization"],
)


@app.middleware("http")
async def limit_request_body_size(request: Request, call_next):
    """Middleware to limit request body size and prevent DoS via large payloads."""
    content_length = request.headers.get("content-length")
    if content_length:
        try:
            if int(content_length) > MAX_REQUEST_BODY_SIZE:
                return JSONResponse(
                    status_code=413,
                    content={"detail": f"Request body too large. Max size: {MAX_REQUEST_BODY_SIZE} bytes"},
                )
        except ValueError:
            pass  # Invalid content-length header, let request proceed
    return await call_next(request)


# =============================================================================
# AUTHENTICATION
# =============================================================================

async def verify_api_key(api_key: Optional[str] = Security(API_KEY_HEADER)) -> Optional[str]:
    """
    Verify API key from header.
    
    Returns the API key if valid, raises HTTPException if invalid.
    """
    if not REQUIRE_AUTH:
        return None  # Auth disabled
    
    if not API_KEY:
        logger.warning("API_KEY not configured but REQUIRE_AUTH=true")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Server authentication not configured",
        )
    
    if not api_key:
        audit_logger.warning("Missing API key in request")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing API key",
            headers={"WWW-Authenticate": "ApiKey"},
        )
    
    # Use constant-time comparison to prevent timing attacks
    if not secrets.compare_digest(api_key, API_KEY):
        audit_logger.warning(f"Invalid API key attempt: {hashlib.sha256(api_key.encode()).hexdigest()[:16]}...")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key",
            headers={"WWW-Authenticate": "ApiKey"},
        )
    
    return api_key


# =============================================================================
# REQUEST CONTEXT MIDDLEWARE
# =============================================================================

@app.middleware("http")
async def add_request_context(request: Request, call_next):
    """Add request ID and client IP to all requests for audit logging (thread-safe)."""
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())[:8])
    client_ip = request.client.host if request.client else "unknown"
    
    # Store in request state for access in endpoints
    request.state.request_id = request_id
    request.state.client_ip = client_ip
    
    # Set context variables (thread-safe, no global state mutation)
    request_id_token = request_id_var.set(request_id)
    client_ip_token = client_ip_var.set(client_ip)
    
    try:
        # Log request
        audit_logger.info(f"Request: {request.method} {request.url.path}")
        
        start_time = time.perf_counter()
        response = await call_next(request)
        duration_ms = (time.perf_counter() - start_time) * 1000
        
        # Add request ID to response headers
        response.headers["X-Request-ID"] = request_id
        
        # Add security headers for HIPAA compliance
        if HIPAA_ENABLED:
            # Use relaxed headers for docs endpoints
            is_docs = request.url.path in ["/docs", "/redoc", "/openapi.json"]
            for header_name, header_value in get_security_headers(is_docs=is_docs).items():
                response.headers[header_name] = header_value
        
        # Log response
        audit_logger.info(f"Response: {response.status_code} ({duration_ms:.2f}ms)")
        
        return response
    finally:
        # Reset context variables
        request_id_var.reset(request_id_token)
        client_ip_var.reset(client_ip_token)

# Global model resources (loaded at startup)
tokenizer: Optional[AutoTokenizer] = None
session: Optional[ort.InferenceSession] = None
label2id: Dict[str, int] = {}
id2label: Dict[int, str] = {}
decision_policy: Dict[str, Any] = {}

# Decision Engine globals (Phase 1: Safety Trifecta)
decision_config: Optional[Any] = None       # DecisionConfigLoader
trigger_engine: Optional[Any] = None        # TaxonomyTriggerEngine
domain_nominator: Optional[Any] = None      # DomainNominator
risk_classifier: Optional[Any] = None       # RiskClassifier
flow_engine: Optional[Any] = None           # FlowEngine (Phase 2)
tenant_manager: Optional[Any] = None       # TenantManager (Phase 3)
journey_orch: Optional[Any] = None         # JourneyOrchestrator (Phase 3)
context_builder: Optional[Any] = None      # PatientContextBuilder (Phase 3)


class PredictRequest(BaseModel):
    """Request schema for /predict endpoint."""
    text: str = Field(..., min_length=1, max_length=MAX_INPUT_CHARS, description="Input text to classify")
    context: Optional[str] = Field(None, max_length=MAX_INPUT_CHARS, description="Optional conversation context")
    
    @field_validator("text", "context", mode="before")
    @classmethod
    def sanitize_input(cls, v: Optional[str]) -> Optional[str]:
        """Sanitize input to prevent injection attacks."""
        if v is None:
            return None
        # Strip HTML/script tags
        v = bleach.clean(v, tags=[], strip=True)
        # Normalize whitespace
        v = " ".join(v.split())
        # Remove null bytes and other control characters
        v = v.replace("\x00", "").replace("\r", "")
        return v
    
    def validate_combined_length(self) -> None:
        """Validate that text + context doesn't exceed combined limit."""
        text_len = len(self.text) if self.text else 0
        context_len = len(self.context) if self.context else 0
        if text_len + context_len > MAX_COMBINED_CHARS:
            raise ValueError(
                f"Combined text and context length ({text_len + context_len}) "
                f"exceeds maximum ({MAX_COMBINED_CHARS})"
            )


class TimingInfo(BaseModel):
    """Timing breakdown for the prediction."""
    tokenize_ms: float
    inference_ms: float
    postprocess_ms: float
    total_ms: float


class TaxonomyMatchDetail(BaseModel):
    """Detail for a single matched taxonomy domain."""
    domain: str
    score: float = Field(..., description="Deterministic match score 0.0-1.0")
    confidence_tier: str = Field(..., description="high/medium/low/none")
    priority: int = Field(0, description="Domain clinical priority")
    is_negated: bool = False
    target_risk_class: Optional[str] = None
    recommended_flow: Optional[str] = None
    trigger_count: int = 0
    matched_phrases: List[str] = Field(default_factory=list)


class PredictResponse(BaseModel):
    """Response schema for /predict endpoint."""
    predicted_label: str
    probabilities: Dict[str, float]
    taxonomy_scores: Dict[str, float] = Field(default_factory=dict, description="Deterministic match scores for all taxonomy domains")
    matched_taxonomy: List[TaxonomyMatchDetail] = Field(default_factory=list, description="Taxonomy domains that matched with details")
    top_taxonomy_domain: Optional[str] = Field(None, description="Highest-scoring taxonomy domain")
    should_escalate: bool
    escalation_probability: float
    confidence: float
    decision_reason: str
    timing: TimingInfo
    model_version: str = "1.0.0"


class DomainNominationResponse(BaseModel):
    """A single domain nomination in the clinical assessment."""
    domain: str
    confidence_tier: str
    priority: int
    target_risk_class: Optional[str] = None
    is_negated: bool = False
    recommended_flow: Optional[str] = None
    trigger_match_count: int = 0


class ClinicalAssessmentResponse(BaseModel):
    """Full clinical risk assessment response for /predict/clinical."""
    # ML classification
    predicted_label: str
    probabilities: Dict[str, float]
    ml_confidence: float
    # Clinical risk assessment
    risk_class: str = Field(..., description="R0/R1/R2/R3 clinical risk tier")
    risk_description: str = Field(..., description="Human-readable risk description")
    primary_domain: Optional[str] = Field(None, description="Primary clinical domain detected")
    requires_911: bool = Field(False, description="Whether patient should be instructed to call 911")
    requires_nurse_transfer: bool = Field(False, description="Whether a nurse transfer is needed")
    hard_escalate: bool = Field(False, description="Whether hard-escalate domain triggered")
    safety_override: bool = Field(False, description="Whether lexical safety signal overrode ML")
    # Domain nominations
    domain_nominations: List[DomainNominationResponse] = Field(default_factory=list)
    # Taxonomy scores (deterministic)
    taxonomy_scores: Dict[str, float] = Field(default_factory=dict, description="Deterministic match scores for all taxonomy domains")
    matched_taxonomy: List[TaxonomyMatchDetail] = Field(default_factory=list, description="Taxonomy domains that matched with details")
    top_taxonomy_domain: Optional[str] = Field(None, description="Highest-scoring taxonomy domain")
    # Flow recommendation
    recommended_flow: Optional[str] = None
    recommended_action: Optional[str] = None
    # Legacy compatibility
    should_escalate: bool
    escalation_probability: float
    decision_reason: str
    # Timing
    timing: TimingInfo
    model_version: str = "1.0.0"
    engine_version: str = "phase1"


class HealthResponse(BaseModel):
    """Response schema for /health endpoint."""
    status: str
    model_loaded: bool
    policy_loaded: bool
    decision_engine_loaded: bool = False
    timestamp: str


def get_onnx_optimization_level(level: str) -> ort.GraphOptimizationLevel:
    """Convert string optimization level to ONNX Runtime enum."""
    levels = {
        "disabled": ort.GraphOptimizationLevel.ORT_DISABLE_ALL,
        "basic": ort.GraphOptimizationLevel.ORT_ENABLE_BASIC,
        "extended": ort.GraphOptimizationLevel.ORT_ENABLE_EXTENDED,
        "all": ort.GraphOptimizationLevel.ORT_ENABLE_ALL,
    }
    return levels.get(level.lower(), ort.GraphOptimizationLevel.ORT_ENABLE_ALL)


def load_model() -> None:
    """Load ONNX model, tokenizer, and decision policy."""
    global tokenizer, session, label2id, id2label, decision_policy, inference_executor
    
    logger.info(f"Loading model from {MODEL_DIR}")
    logger.info(f"ONNX config: intra_threads={ONNX_INTRA_OP_THREADS}, inter_threads={ONNX_INTER_OP_THREADS}, opt_level={ONNX_GRAPH_OPT_LEVEL}")
    
    # Initialize thread pool for async inference
    inference_executor = ThreadPoolExecutor(
        max_workers=INFERENCE_WORKERS,
        thread_name_prefix="inference_worker"
    )
    logger.info(f"Inference thread pool created with {INFERENCE_WORKERS} workers")
    
    # Load tokenizer
    tokenizer = AutoTokenizer.from_pretrained(MODEL_DIR)
    logger.info("Tokenizer loaded")
    
    # Load ONNX model
    model_path = MODEL_DIR / "model.onnx"
    if not model_path.exists():
        raise FileNotFoundError(f"ONNX model not found: {model_path}")
    
    # Configure session options with environment-based settings
    sess_options = ort.SessionOptions()
    sess_options.graph_optimization_level = get_onnx_optimization_level(ONNX_GRAPH_OPT_LEVEL)
    sess_options.intra_op_num_threads = ONNX_INTRA_OP_THREADS
    sess_options.inter_op_num_threads = ONNX_INTER_OP_THREADS
    
    # Enable memory optimization for better scalability
    sess_options.enable_mem_pattern = True
    sess_options.enable_cpu_mem_arena = True
    
    session = ort.InferenceSession(
        str(model_path),
        sess_options,
        providers=["CPUExecutionProvider"]
    )
    logger.info("ONNX session created with optimized settings")
    
    # Check quantization metadata - warn if model is not actually quantized
    quant_meta_path = MODEL_DIR / "quantization_metadata.json"
    if quant_meta_path.exists():
        with open(quant_meta_path, "r", encoding="utf-8") as f:
            quant_meta = json.load(f)
        if not quant_meta.get("is_quantized", True):
            logger.warning(
                f"MODEL NOT QUANTIZED: The model in {MODEL_DIR} is an unquantized FP32 model "
                f"despite being in a directory that implies INT8 quantization. "
                f"Inference will be slower than expected. Re-run: python quantize_onnx.py"
            )
    
    # Load label mappings
    mappings_path = MODEL_DIR / "label_mappings.json"
    if mappings_path.exists():
        with open(mappings_path, "r", encoding="utf-8") as f:
            mappings = json.load(f)
        label2id = mappings["label2id"]
        id2label = {int(k): v for k, v in mappings["id2label"].items()}
        logger.info(f"Loaded {len(label2id)} labels")
    else:
        logger.warning("Label mappings not found, using defaults")
    
    # Load decision policy
    if POLICY_PATH.exists():
        with open(POLICY_PATH, "r", encoding="utf-8") as f:
            decision_policy = json.load(f)
        logger.info("Decision policy loaded")
        
        # Runtime safety guardrail: warn if calibrated recall is below minimum
        SAFETY_MIN_RECALL = 0.85
        cal_metrics = decision_policy.get("calibration_metrics", {})
        policy_recall = cal_metrics.get("recall", None)
        if policy_recall is not None and policy_recall < SAFETY_MIN_RECALL:
            logger.warning(
                f"SAFETY WARNING: Decision policy recall ({policy_recall:.3f}) is below "
                f"the safety minimum ({SAFETY_MIN_RECALL:.3f}). "
                f"{(1 - policy_recall) * 100:.1f}% of escalation cases may be missed. "
                f"Recalibrate with: python calibrate.py --max_length 64"
            )
    else:
        logger.warning(f"Decision policy not found at {POLICY_PATH}, using defaults")
        decision_policy = {
            "escalation_policy": {
                "escalation_label": DEFAULT_ESCALATION_LABEL,
                "escalation_threshold": DEFAULT_ESCALATION_THRESHOLD,
            },
            "confidence_policy": {
                "min_confidence": DEFAULT_MIN_CONFIDENCE,
                "fallback_label": DEFAULT_FALLBACK_LABEL,
            },
        }
    
    # Initialize Decision Engine (Phase 1: Safety Trifecta)
    global decision_config, trigger_engine, domain_nominator, risk_classifier, flow_engine, tenant_manager, journey_orch, context_builder
    if DECISION_ENGINE_AVAILABLE:
        decision_dir = Path(__file__).resolve().parent.parent / "decision"
        if decision_dir.is_dir():
            try:
                decision_config = DecisionConfigLoader(decision_dir)
                decision_config.load_all()
                # Load additional taxonomy domains from list/taxonomy/
                list_taxonomy_dir = Path(__file__).resolve().parent.parent / "list" / "taxonomy"
                if list_taxonomy_dir.is_dir():
                    decision_config.load_additional_taxonomy(list_taxonomy_dir)
                trigger_engine = TaxonomyTriggerEngine(decision_config)
                domain_nominator = DomainNominator(decision_config, trigger_engine)
                risk_classifier = RiskClassifier(decision_config)
                flow_engine = FlowEngine(decision_config)
                tenant_manager = TenantManager(decision_config)
                journey_orch = JourneyOrchestrator(decision_config, tenant_manager)
                context_builder = PatientContextBuilder(decision_config)
                logger.info(
                    "Decision engine initialized: %d domains, %d risk rules, %d suppression rules, "
                    "%d tenants, %d journeys, flow engine ready",
                    len(decision_config.domain_names),
                    len(decision_config.risk_escalation_rules),
                    len(decision_config.domain_suppression_rules),
                    len(decision_config.tenants),
                    len(decision_config.journeys),
                )
            except Exception as e:
                logger.error("Failed to initialize decision engine: %s", e)
                decision_config = None
                trigger_engine = None
                domain_nominator = None
                risk_classifier = None
        else:
            logger.info("Decision config directory not found at %s, engine disabled", decision_dir)


def softmax(x: np.ndarray) -> np.ndarray:
    """Compute softmax probabilities from logits."""
    x_shifted = x - np.max(x, axis=-1, keepdims=True)
    exp_x = np.exp(x_shifted)
    return exp_x / np.sum(exp_x, axis=-1, keepdims=True)


# Safety keywords that ALWAYS trigger escalation regardless of model output
# Each entry: (pattern, description, requires_negation_check)
SAFETY_ESCALATION_PATTERNS = [
    # Mental health - ALWAYS check for negation
    (re.compile(r"\b(suicide|suicidal)\b", re.IGNORECASE), "suicide_mention", True),
    (re.compile(r"\b(kill myself|end my life|want to die|kill me)\b", re.IGNORECASE), "self_harm_intent", True),
    (re.compile(r"\b(hurt myself|self[- ]?harm|cutting myself)\b", re.IGNORECASE), "self_harm", True),
    (re.compile(r"\b(overdose|od['']?d)\b", re.IGNORECASE), "overdose", True),
    # Breathing - check for negation (e.g., "I'm not having trouble breathing")
    (re.compile(r"\b(can'?t breathe|cannot breathe|not breathing|stop breathing)\b", re.IGNORECASE), "breathing_emergency", False),  # These phrases are already negative
    (re.compile(r"\b(difficulty breathing|trouble breathing|hard to breathe)\b", re.IGNORECASE), "breathing_difficulty", True),
    # Cardiac - check for negation
    (re.compile(r"\b(chest\s+(pain|hurts?|pressure|tightness|burning|discomfort)|pain\s+in\s+(my\s+)?chest|heart\s+pain)\b", re.IGNORECASE), "chest_pain", True),
    (re.compile(r"\b(heart attack|cardiac arrest)\b", re.IGNORECASE), "cardiac_emergency", True),
    (re.compile(r"\b(stroke)\b", re.IGNORECASE), "stroke", True),
    # Consciousness - check for negation
    (re.compile(r"\b(unconscious|unresponsive|passed out)\b", re.IGNORECASE), "unconscious", True),
    (re.compile(r"\b(seizure|convulsion)\b", re.IGNORECASE), "seizure", True),
    # Bleeding - check for negation
    (re.compile(r"\b(severe bleeding|bleeding (heavily|out|profusely)|hemorrhage|hemorrhaging)\b", re.IGNORECASE), "severe_bleeding", True),
    # Allergic reaction - check for negation
    (re.compile(r"\b(anaphylaxis|anaphylactic|severe allergic)\b", re.IGNORECASE), "anaphylaxis", True),
    (re.compile(r"\b(throat.*(closing|swelling))\b", re.IGNORECASE), "throat_swelling", True),
]

# Negation patterns to check before safety keywords
NEGATION_PATTERNS = [
    re.compile(r"\b(not|no|don'?t|doesn'?t|didn'?t|haven'?t|hasn'?t|hadn'?t|won'?t|wouldn'?t|can'?t|cannot|never|without|ain'?t)\b", re.IGNORECASE),
]

# Context window for negation check (number of words before keyword)
NEGATION_WINDOW_SIZE = 5

# Historical/third-party patterns that indicate non-emergency context
HISTORICAL_PATTERNS = [
    re.compile(r"\b(my (friend|mother|father|brother|sister|husband|wife|son|daughter|grandma|grandpa|neighbor|coworker))\b", re.IGNORECASE),
    re.compile(r"\b(last (year|month|week|time)|years? ago|previously|in the past|used to|history of)\b", re.IGNORECASE),
    re.compile(r"\b(worried about|afraid of|scared of|fear of|thinking about whether)\b", re.IGNORECASE),
]

# Active-urgency patterns that OVERRIDE historical/proxy suppression.
# When a proxy reference ("my mother") co-occurs with these acute signals,
# the report should be treated as authoritative — do NOT suppress.
PROXY_ACTIVE_URGENCY_PATTERNS = [
    re.compile(r"\b(is\s+having|are\s+having|has\s+been\s+having)\b", re.IGNORECASE),
    re.compile(r"\b(right\s+now|at\s+this\s+moment|currently|presently)\b", re.IGNORECASE),
    re.compile(r"\b(can'?t\s+breathe|cannot\s+breathe|not\s+breathing|stopped\s+breathing)\b", re.IGNORECASE),
    re.compile(r"\b(is\s+(bleeding|seizing|convulsing|choking|vomiting|fainting|unconscious|unresponsive))\b", re.IGNORECASE),
    re.compile(r"\b(just\s+(collapsed|fainted|passed\s+out|fell|overdosed|seized))\b", re.IGNORECASE),
    re.compile(r"\b(needs?\s+help|call\s+911|emergency|ambulance)\b", re.IGNORECASE),
    re.compile(r"\b(what\s+should\s+(i|we|she|he|they)\s+do)\b", re.IGNORECASE),
    re.compile(r"\b(is\s+(in\s+pain|hurting|suffering|getting\s+worse))\b", re.IGNORECASE),
    re.compile(r"\b(says?\s+(she|he|they)\s+(can'?t|has|is|feels?))\b", re.IGNORECASE),
]


_POST_NEGATION_PATTERNS = [
    re.compile(r"\b(free|resolved|gone|cleared|negative|ruled\s+out|improving|controlled|stable)\b", re.IGNORECASE),
]


def _check_negation_in_window(text: str, match_start: int, window_size: int = NEGATION_WINDOW_SIZE, match_end: int = 0) -> bool:
    """
    Check if there's a negation word within window_size words before the match,
    OR a post-negation/resolution word immediately after the match.
    
    Args:
        text: Full text to check
        match_start: Character position where the safety keyword starts
        window_size: Number of words to look back for negation
        match_end: Character position where the safety keyword ends (for post-check)
        
    Returns:
        True if negation found in window, False otherwise
    """
    # Get text before the match
    text_before = text[:match_start].lower()
    
    # Split into words and get the last N words
    words_before = text_before.split()
    window_words = words_before[-window_size:] if len(words_before) >= window_size else words_before
    window_text = " ".join(window_words)
    
    # Check for negation patterns in the window BEFORE the keyword
    for neg_pattern in NEGATION_PATTERNS:
        if neg_pattern.search(window_text):
            return True

    # Check for post-negation words AFTER the keyword (e.g., "seizure free")
    if match_end > 0:
        text_after = text[match_end:match_end + 30].lower().strip()
        for post_pattern in _POST_NEGATION_PATTERNS:
            if post_pattern.search(text_after):
                return True
    
    return False


def _check_historical_context(text: str, match_start: int) -> bool:
    """
    Check if the context suggests historical/third-party reference.
    
    Returns True (suppress escalation) ONLY if historical/proxy context is
    found AND there are no active-urgency signals indicating the report is
    about a current emergency. Proxy reports with acute symptoms (e.g.,
    "my mother is having chest pain") are treated as authoritative.
    
    Args:
        text: Full text to check
        match_start: Character position where the safety keyword starts
        
    Returns:
        True if historical/third-party context found WITHOUT active urgency,
        False otherwise (i.e., DO escalate)
    """
    # Check the sentence or nearby text for historical patterns
    # Get text around the match (100 chars before)
    context_start = max(0, match_start - 100)
    context_text = text[context_start:match_start].lower()
    
    has_historical = False
    for pattern in HISTORICAL_PATTERNS:
        if pattern.search(context_text):
            has_historical = True
            break
    
    if not has_historical:
        return False
    
    # Historical/proxy pattern found — but check if there are active-urgency
    # signals in the FULL text that indicate a current emergency.
    # E.g., "my mother is having chest pain" should NOT be suppressed.
    text_lower = text.lower()
    for urgency_pattern in PROXY_ACTIVE_URGENCY_PATTERNS:
        if urgency_pattern.search(text_lower):
            logger.info(
                "Historical/proxy context found but active-urgency signal "
                "overrides suppression — treating as authoritative report"
            )
            return False
    
    return True


def check_safety_keywords(text: str) -> Tuple[bool, Optional[str]]:
    """
    Check text for critical safety keywords with negation awareness.
    
    This is a rule-based safety net that triggers escalation regardless of
    model predictions. Critical for patient safety in clinical settings.
    
    Negation-aware: Will NOT trigger on phrases like:
    - "I'm not suicidal"
    - "I don't have chest pain"
    - "My friend had a heart attack last year"
    
    Returns:
        Tuple of (should_escalate, matched_pattern_description)
    """
    text_lower = text.lower()
    
    for pattern, description, check_negation in SAFETY_ESCALATION_PATTERNS:
        match = pattern.search(text_lower)
        if match:
            match_start = match.start()
            
            # Check for negation if required
            if check_negation:
                if _check_negation_in_window(text, match_start, match_end=match.end()):
                    # Negation found - don't escalate but log for review
                    logger.info(f"Safety keyword '{match.group()}' found but negated - not escalating")
                    continue
                
                # Check for historical/third-party context
                if _check_historical_context(text, match_start):
                    logger.info(f"Safety keyword '{match.group()}' found in historical context - not escalating")
                    continue
            
            # No negation or historical context - escalate
            return True, f"safety_keyword_match: '{match.group()}' ({description})"
    
    return False, None


def apply_decision_policy(
    probabilities: np.ndarray,
    argmax_label: str,
    input_text: Optional[str] = None
) -> Tuple[str, bool, str]:
    """
    Apply decision policy to determine final label and escalation status.
    
    Includes safety keyword fallback for critical patient safety.
    
    Returns:
        Tuple of (final_label, should_escalate, decision_reason)
    """
    # SAFETY FIRST: Check for critical keywords before model decision
    if input_text:
        keyword_escalate, keyword_reason = check_safety_keywords(input_text)
        if keyword_escalate:
            escalation_label = decision_policy.get("escalation_policy", {}).get("escalation_label", "ESCALATION")
            return escalation_label, True, f"SAFETY_OVERRIDE: {keyword_reason}"
    
    escalation_config = decision_policy.get("escalation_policy", {})
    confidence_config = decision_policy.get("confidence_policy", {})
    
    escalation_label = escalation_config.get("escalation_label", "ESCALATION")
    escalation_threshold = escalation_config.get("escalation_threshold", 0.5)
    min_confidence = confidence_config.get("min_confidence", 0.3)
    fallback_label = confidence_config.get("fallback_label", "OTHER")
    
    # Get escalation probability
    escalation_id = label2id.get(escalation_label, 0)
    escalation_prob = float(probabilities[escalation_id])
    
    # Get max probability (confidence)
    max_prob = float(np.max(probabilities))
    
    # Apply decision logic
    # Rule 1: If escalation probability exceeds threshold, always escalate
    if escalation_prob >= escalation_threshold:
        return escalation_label, True, f"P({escalation_label})={escalation_prob:.3f} >= threshold={escalation_threshold}"
    
    # Rule 2: If confidence is too low, route to fallback
    if max_prob < min_confidence:
        return fallback_label, False, f"Low confidence ({max_prob:.3f} < {min_confidence}), routing to {fallback_label}"
    
    # Rule 3: Use argmax prediction
    return argmax_label, False, f"Standard prediction with confidence={max_prob:.3f}"


def shutdown_cleanup() -> None:
    """Clean up resources on shutdown."""
    global inference_executor
    if inference_executor:
        logger.info("Shutting down inference thread pool...")
        inference_executor.shutdown(wait=True, cancel_futures=False)
        logger.info("Inference thread pool shut down")


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifespan context manager for startup and shutdown events."""
    # Startup
    try:
        load_model()
        logger.info("Service ready")
    except Exception as e:
        logger.error(f"Failed to load model: {e}")
        raise
    
    yield  # Application runs here
    
    # Shutdown
    shutdown_cleanup()
    logger.info("Service shutdown complete")


# Register lifespan with app
app.router.lifespan_context = lifespan


# Graceful shutdown signal handler
def handle_shutdown_signal(signum, frame):
    """Handle shutdown signals gracefully."""
    logger.info(f"Received signal {signum}, initiating graceful shutdown...")
    shutdown_cleanup()
    sys.exit(0)


# Register signal handlers (Unix-like systems)
if hasattr(signal, 'SIGTERM'):
    signal.signal(signal.SIGTERM, handle_shutdown_signal)
if hasattr(signal, 'SIGINT'):
    signal.signal(signal.SIGINT, handle_shutdown_signal)


def run_inference_sync(input_ids: np.ndarray, attention_mask: np.ndarray, token_type_ids: Optional[np.ndarray]) -> np.ndarray:
    """
    Run ONNX inference synchronously (called from thread pool).
    
    This function is CPU-bound and should be run in a thread pool
    to avoid blocking the async event loop.
    """
    ort_inputs = {
        "input_ids": input_ids,
        "attention_mask": attention_mask,
    }
    
    # Add token_type_ids if model expects it
    input_names = [inp.name for inp in session.get_inputs()]
    if "token_type_ids" in input_names and token_type_ids is not None:
        ort_inputs["token_type_ids"] = token_type_ids
    
    outputs = session.run(None, ort_inputs)
    return outputs[0][0]  # Return logits for first sample


async def run_inference_async(input_ids: np.ndarray, attention_mask: np.ndarray, token_type_ids: Optional[np.ndarray]) -> np.ndarray:
    """
    Run ONNX inference asynchronously using thread pool with timeout.
    
    This prevents blocking the async event loop during CPU-bound inference.
    Raises asyncio.TimeoutError if inference takes longer than INFERENCE_TIMEOUT_SECONDS.
    """
    loop = asyncio.get_running_loop()
    try:
        return await asyncio.wait_for(
            loop.run_in_executor(
                inference_executor,
                run_inference_sync,
                input_ids,
                attention_mask,
                token_type_ids
            ),
            timeout=INFERENCE_TIMEOUT_SECONDS
        )
    except asyncio.TimeoutError:
        logger.error(f"Inference timed out after {INFERENCE_TIMEOUT_SECONDS}s")
        raise


# Health check cache to prevent DoS via repeated inference
_health_check_cache: Dict[str, Any] = {"status": None, "timestamp": 0.0}
_health_check_cache_lock = threading.Lock()
HEALTH_CHECK_CACHE_TTL = 5.0  # Cache health status for 5 seconds


@app.get("/health", response_model=HealthResponse)
@limiter.limit("30/minute")
async def health_check(request: Request):
    """Health check endpoint (no auth required for load balancer probes)."""
    current_time = time.time()
    
    # Return cached result if within TTL to prevent DoS via repeated inference
    with _health_check_cache_lock:
        if (
            _health_check_cache["status"] is not None
            and current_time - _health_check_cache["timestamp"] < HEALTH_CHECK_CACHE_TTL
        ):
            cached = _health_check_cache["status"]
            return HealthResponse(
                status=cached["status"],
                model_loaded=cached["model_loaded"],
                policy_loaded=cached["policy_loaded"],
                decision_engine_loaded=cached.get("decision_engine_loaded", False),
                timestamp=datetime.now(timezone.utc).isoformat(),
            )
    
    model_healthy = False
    if session is not None and tokenizer is not None:
        try:
            # Quick validation that model can run inference
            test_inputs = tokenizer("test", return_tensors="np", max_length=8, truncation=True, padding="max_length")
            input_dict = {"input_ids": test_inputs["input_ids"], "attention_mask": test_inputs["attention_mask"]}
            input_names = [inp.name for inp in session.get_inputs()]
            if "token_type_ids" in input_names:
                input_dict["token_type_ids"] = np.zeros_like(test_inputs["input_ids"])
            session.run(None, input_dict)
            model_healthy = True
        except Exception:
            model_healthy = False
    
    engine_loaded = DECISION_ENGINE_AVAILABLE and domain_nominator is not None
    
    # Cache the result (thread-safe)
    with _health_check_cache_lock:
        _health_check_cache["status"] = {
            "status": "healthy" if model_healthy else "unhealthy",
            "model_loaded": model_healthy,
            "policy_loaded": bool(decision_policy),
            "decision_engine_loaded": engine_loaded,
        }
        _health_check_cache["timestamp"] = current_time
    
    return HealthResponse(
        status="healthy" if model_healthy else "unhealthy",
        model_loaded=model_healthy,
        policy_loaded=bool(decision_policy),
        decision_engine_loaded=engine_loaded,
        timestamp=datetime.now(timezone.utc).isoformat(),
    )


@app.post("/predict", response_model=PredictResponse)
@limiter.limit(RATE_LIMIT)
async def predict(
    request: Request,
    predict_request: PredictRequest,
    api_key: str = Depends(verify_api_key),
):
    """
    Classify medical intent from input text.
    
    Applies decision policy for escalation thresholding.
    Requires API key authentication.
    
    Uses async inference to prevent blocking the event loop under load.
    """
    if session is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    # Validate combined length
    try:
        predict_request.validate_combined_length()
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    
    total_start = time.perf_counter()
    
    # Prepare input text — run temporal preprocessor first
    text = predict_request.text
    temporal_signal = None
    if TEMPORAL_PREPROCESSOR_AVAILABLE:
        temporal_signal = preprocess_temporal(text, predict_request.context)
        text = temporal_signal.processed_text
    
    # Tokenize (fast enough to run synchronously)
    # When context is provided, use BERT's native text_pair so that:
    #   1. Text (current utterance) is Segment A — always fully preserved
    #   2. Context is Segment B — truncated first if combined length exceeds max_length
    # truncation="only_second" ensures the patient's current text is NEVER truncated
    tokenize_start = time.perf_counter()
    if predict_request.context:
        inputs = tokenizer(
            text,
            text_pair=predict_request.context,
            padding="max_length",
            truncation="only_second",
            max_length=MAX_LENGTH,
            return_tensors="np"
        )
    else:
        inputs = tokenizer(
            text,
            padding="max_length",
            truncation=True,
            max_length=MAX_LENGTH,
            return_tensors="np"
        )
    tokenize_time = (time.perf_counter() - tokenize_start) * 1000
    
    # Prepare ONNX inputs
    input_ids = inputs["input_ids"].astype(np.int64)
    attention_mask = inputs["attention_mask"].astype(np.int64)
    
    # Prepare token_type_ids if needed
    token_type_ids = None
    input_names = [inp.name for inp in session.get_inputs()]
    if "token_type_ids" in input_names:
        if "token_type_ids" in inputs:
            token_type_ids = inputs["token_type_ids"].astype(np.int64)
        else:
            token_type_ids = np.zeros_like(input_ids, dtype=np.int64)
    
    # Run inference asynchronously (prevents blocking event loop)
    inference_start = time.perf_counter()
    logits = await run_inference_async(input_ids, attention_mask, token_type_ids)
    inference_time = (time.perf_counter() - inference_start) * 1000
    
    # Postprocess
    postprocess_start = time.perf_counter()
    probabilities = softmax(logits)
    
    # Get argmax prediction
    argmax_id = int(np.argmax(probabilities))
    argmax_label = id2label.get(argmax_id, "UNKNOWN")
    
    # Apply decision policy (includes safety keyword fallback)
    # Use original text (not context-concatenated) for safety keyword checks
    # to avoid false escalation from keywords in historical context
    final_label, should_escalate, decision_reason = apply_decision_policy(
        probabilities, argmax_label, input_text=predict_request.text
    )
    
    # Temporal preprocessor override: if preprocessor detected active emergency
    # but model/policy didn't escalate, force escalation (safety net)
    if temporal_signal and temporal_signal.is_active_emergency and not should_escalate:
        escalation_label_name = decision_policy.get("escalation_policy", {}).get("escalation_label", "ESCALATION")
        final_label = escalation_label_name
        should_escalate = True
        decision_reason = f"TEMPORAL_OVERRIDE: {temporal_signal.reason}"
    
    # Build probability dict
    prob_dict = {
        id2label.get(i, f"LABEL_{i}"): float(p)
        for i, p in enumerate(probabilities)
    }
    
    # Get escalation probability
    escalation_label = decision_policy.get("escalation_policy", {}).get("escalation_label", "ESCALATION")
    escalation_prob = prob_dict.get(escalation_label, 0.0)
    
    postprocess_time = (time.perf_counter() - postprocess_start) * 1000
    total_time = (time.perf_counter() - total_start) * 1000
    
    # Record metrics
    if ENABLE_METRICS:
        metrics.record_request(
            success=True,
            inference_ms=inference_time,
            tokenize_ms=tokenize_time,
            escalated=should_escalate
        )
    
    # Taxonomy scoring (deterministic)
    tax_scores: Dict[str, float] = {}
    matched_tax: List[TaxonomyMatchDetail] = []
    top_tax_domain: Optional[str] = None
    if TAXONOMY_SCORER_AVAILABLE and trigger_engine and decision_config:
        try:
            tax_result = score_taxonomy(predict_request.text, trigger_engine, decision_config)
            tax_scores = tax_result.scores
            top_tax_domain = tax_result.top_domain
            matched_tax = [
                TaxonomyMatchDetail(
                    domain=d.domain,
                    score=round(d.score, 4),
                    confidence_tier=d.confidence_tier,
                    priority=d.priority,
                    is_negated=d.is_negated,
                    target_risk_class=d.target_risk_class,
                    recommended_flow=d.recommended_flow,
                    trigger_count=d.trigger_count,
                    matched_phrases=d.matched_phrases,
                )
                for d in tax_result.matched_domains
            ]

            # SAFETY NET: If any matched taxonomy domain is a hard-escalate
            # R3 domain (not negated), force escalation even when ML disagrees.
            # For temporally resolved cases: still escalate but note follow-up
            # (past R3 symptoms warrant clinical assessment, not suppression).
            is_temporally_resolved = (
                temporal_signal
                and temporal_signal.temporal_tag == "resolved"
            )
            if decision_config:
                hard_esc_domains = decision_config.hard_escalate_domains
                for d in tax_result.matched_domains:
                    if d.domain in hard_esc_domains and not d.is_negated and d.score > 0:
                        if not should_escalate:
                            escalation_label_name = decision_policy.get(
                                "escalation_policy", {}
                            ).get("escalation_label", "ESCALATION")
                            final_label = escalation_label_name
                            should_escalate = True
                            if is_temporally_resolved:
                                decision_reason = (
                                    f"TAXONOMY_FOLLOW_UP: {d.domain} "
                                    f"(resolved but R3 domain warrants clinical follow-up)"
                                )
                            else:
                                decision_reason = (
                                    f"TAXONOMY_ESCALATION: {d.domain} "
                                    f"(target_risk={d.target_risk_class}, "
                                    f"confidence={d.confidence_tier})"
                                )
                            logger.warning(
                                "Taxonomy safety override on /predict: "
                                "domain=%s risk=%s ML_label=%s temporal=%s",
                                d.domain, d.target_risk_class, argmax_label,
                                "resolved" if is_temporally_resolved else "active",
                            )
                        break
        except Exception as e:
            logger.warning("Taxonomy scoring failed: %s", e)

    return PredictResponse(
        predicted_label=final_label,
        probabilities=prob_dict,
        taxonomy_scores=tax_scores,
        matched_taxonomy=matched_tax,
        top_taxonomy_domain=top_tax_domain,
        should_escalate=should_escalate,
        escalation_probability=escalation_prob,
        confidence=float(np.max(probabilities)),
        decision_reason=decision_reason,
        timing=TimingInfo(
            tokenize_ms=round(tokenize_time, 3),
            inference_ms=round(inference_time, 3),
            postprocess_ms=round(postprocess_time, 3),
            total_ms=round(total_time, 3),
        ),
    )


@app.post("/predict/clinical", response_model=ClinicalAssessmentResponse, tags=["Clinical"])
@limiter.limit(RATE_LIMIT)
async def predict_clinical(
    request: Request,
    predict_request: PredictRequest,
    api_key: str = Depends(verify_api_key),
):
    """
    Clinical risk assessment endpoint.

    Combines DriveHealthBERT ML classification with taxonomy-based domain nomination
    and 4-tier risk classification (R0/R1/R2/R3).

    Returns the full clinical assessment including:
    - ML predicted label and probabilities
    - Primary clinical domain (e.g., chest_pain, suicidal_ideation)
    - Risk class: R0 (no concern) → R3 (call 911)
    - Domain nominations ranked by acuity
    - Recommended conversation flow
    - Safety override and hard-escalate flags
    """
    if session is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    if not DECISION_ENGINE_AVAILABLE or domain_nominator is None:
        raise HTTPException(
            status_code=503,
            detail="Decision engine not loaded. Ensure decision/ config directory exists.",
        )

    try:
        predict_request.validate_combined_length()
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

    total_start = time.perf_counter()

    # Prepare input text — run temporal preprocessor first
    text = predict_request.text
    temporal_signal = None
    if TEMPORAL_PREPROCESSOR_AVAILABLE:
        temporal_signal = preprocess_temporal(text, predict_request.context)
        text = temporal_signal.processed_text

    # Tokenize: text as Segment A (never truncated), context as Segment B (truncated first)
    tokenize_start = time.perf_counter()
    if predict_request.context:
        inputs = tokenizer(
            text,
            text_pair=predict_request.context,
            padding="max_length",
            truncation="only_second",
            max_length=MAX_LENGTH,
            return_tensors="np",
        )
    else:
        inputs = tokenizer(
            text,
            padding="max_length",
            truncation=True,
            max_length=MAX_LENGTH,
            return_tensors="np",
        )
    tokenize_time = (time.perf_counter() - tokenize_start) * 1000

    # Prepare ONNX inputs
    input_ids = inputs["input_ids"].astype(np.int64)
    attention_mask = inputs["attention_mask"].astype(np.int64)
    token_type_ids = None
    input_names = [inp.name for inp in session.get_inputs()]
    if "token_type_ids" in input_names:
        if "token_type_ids" in inputs:
            token_type_ids = inputs["token_type_ids"].astype(np.int64)
        else:
            token_type_ids = np.zeros_like(input_ids, dtype=np.int64)

    # Run ML inference
    inference_start = time.perf_counter()
    logits = await run_inference_async(input_ids, attention_mask, token_type_ids)
    inference_time = (time.perf_counter() - inference_start) * 1000

    # Postprocess ML output
    postprocess_start = time.perf_counter()
    probabilities = softmax(logits)
    argmax_id = int(np.argmax(probabilities))
    argmax_label = id2label.get(argmax_id, "UNKNOWN")
    ml_confidence = float(np.max(probabilities))

    prob_dict = {
        id2label.get(i, f"LABEL_{i}"): float(p)
        for i, p in enumerate(probabilities)
    }

    escalation_label = decision_policy.get("escalation_policy", {}).get(
        "escalation_label", "ESCALATION"
    )
    escalation_prob = prob_dict.get(escalation_label, 0.0)

    # Run Decision Engine — use ORIGINAL text (not context-concatenated)
    nominations = domain_nominator.nominate(
        text=predict_request.text,
        ml_label=argmax_label,
        ml_probabilities=prob_dict,
    )
    assessment = risk_classifier.assess(
        nominations=nominations,
        slot_values={},  # Phase 2: filled by flow engine
        patient_context=None,  # Phase 3: filled by FHIR loader
        ml_label=argmax_label,
        ml_confidence=ml_confidence,
    )

    # Determine final label and escalation (merging ML + decision engine)
    if assessment.risk_class >= RiskClass.R2:
        should_escalate = True
        final_label = escalation_label
        if assessment.hard_escalate:
            decision_reason = f"HARD_ESCALATE: {assessment.primary_domain} → {assessment.risk_class.value}"
        elif assessment.safety_override:
            decision_reason = f"SAFETY_OVERRIDE: {assessment.primary_domain} → {assessment.risk_class.value}"
        else:
            decision_reason = (
                f"CLINICAL_RISK: {assessment.primary_domain} → "
                f"{assessment.risk_class.value} (ML={argmax_label} {ml_confidence:.3f})"
            )
    else:
        # Use legacy decision policy for non-escalation
        final_label, should_escalate, decision_reason = apply_decision_policy(
            probabilities, argmax_label, input_text=predict_request.text
        )
        # Belt + suspenders: if legacy policy escalated but engine says R0/R1,
        # upgrade risk_class to R2 minimum. Never suppress an escalation.
        if should_escalate and assessment.risk_class < RiskClass.R2:
            assessment = RiskAssessment(
                risk_class=RiskClass.R2,
                primary_domain=assessment.primary_domain,
                domain_nominations=assessment.domain_nominations,
                matched_rules=assessment.matched_rules,
                suppressions=assessment.suppressions,
                hard_escalate=assessment.hard_escalate,
                safety_override=True,
                ml_label=assessment.ml_label,
                ml_confidence=assessment.ml_confidence,
                recommended_flow="handoff",
                recommended_action=assessment.recommended_action,
            )

    # Temporal preprocessor override for /predict/clinical
    if temporal_signal and temporal_signal.is_active_emergency and not should_escalate:
        should_escalate = True
        final_label = escalation_label
        decision_reason = f"TEMPORAL_OVERRIDE: {temporal_signal.reason}"
        if assessment.risk_class < RiskClass.R2:
            assessment = RiskAssessment(
                risk_class=RiskClass.R2,
                primary_domain=assessment.primary_domain,
                domain_nominations=assessment.domain_nominations,
                matched_rules=assessment.matched_rules,
                suppressions=assessment.suppressions,
                hard_escalate=assessment.hard_escalate,
                safety_override=True,
                ml_label=assessment.ml_label,
                ml_confidence=assessment.ml_confidence,
                recommended_flow="handoff",
                recommended_action=assessment.recommended_action,
            )

    # CONTRASTIVE REVERSAL OVERRIDE: When temporal preprocessor detects a
    # BUT-clause reversal (e.g., "chest pain went away but it's back"), the
    # domain nomination may have negated the R3 domain (e.g., "chest pain went
    # away" matches negation patterns). Override negation for R3 domains when
    # temporal analysis confirms the condition is currently active.
    if (
        temporal_signal
        and temporal_signal.is_active_emergency
        and any(
            r.startswith("but_clause_reversal") or r == "contrastive_post_active"
            for r in temporal_signal.matched_rules
        )
        and assessment.risk_class < RiskClass.R3
    ):
        # Check if any R3 domain was nominated but negated
        override_domain = None
        if decision_config:
            hard_esc_domains = decision_config.hard_escalate_domains
            for nom in nominations:
                if nom.domain in hard_esc_domains and nom.is_negated:
                    override_domain = nom.domain
                    break
        if override_domain:
            should_escalate = True
            final_label = escalation_label
            decision_reason = (
                f"CONTRASTIVE_OVERRIDE: {override_domain} was negated but "
                f"BUT-clause reversal confirms active condition — R3"
            )
            assessment = RiskAssessment(
                risk_class=RiskClass.R3,
                primary_domain=override_domain,
                domain_nominations=assessment.domain_nominations,
                matched_rules=assessment.matched_rules,
                suppressions=assessment.suppressions,
                hard_escalate=True,
                safety_override=True,
                ml_label=assessment.ml_label,
                ml_confidence=assessment.ml_confidence,
                recommended_flow="escalate_r3",
                recommended_action=None,
            )
            logger.warning(
                "Contrastive reversal override: %s negation overridden, "
                "temporal confirms active — R3",
                override_domain,
            )

    # Temporal DE-ESCALATION: If the temporal preprocessor detected a clearly
    # resolved / past condition AND the escalation came from taxonomy hard-escalate,
    # downgrade from R3 to R2 (clinical follow-up), NOT to R1.
    #
    # SAFETY RATIONALE: Even resolved R3 symptoms warrant clinical follow-up.
    # "I had chest pain yesterday but I'm fine" is STILL a cardiac red flag.
    # R2 = warm transfer to nurse for clinical assessment. R1 would suppress.
    # Proxy reports are treated as authoritative — caregivers speak for patients.
    if (
        temporal_signal
        and temporal_signal.temporal_tag == "resolved"
        and should_escalate
        and assessment.hard_escalate
    ):
        # Downgrade to R2 (clinical follow-up), keep should_escalate=True
        decision_reason = (
            f"TEMPORAL_FOLLOW_UP: {assessment.primary_domain} detected, "
            f"condition appears resolved ({temporal_signal.reason}) "
            f"— downgraded to R2 clinical follow-up"
        )
        assessment = RiskAssessment(
            risk_class=RiskClass.R2,
            primary_domain=assessment.primary_domain,
            domain_nominations=assessment.domain_nominations,
            matched_rules=assessment.matched_rules,
            suppressions=assessment.suppressions,
            hard_escalate=False,
            safety_override=True,
            ml_label=assessment.ml_label,
            ml_confidence=assessment.ml_confidence,
            recommended_flow="clinical_follow_up",
            recommended_action="Warm transfer to nurse — past R3 symptom warrants clinical assessment",
        )
        logger.info(
            "Temporal de-escalation: %s resolved, downgrading from R3 to R2 (clinical follow-up)",
            assessment.primary_domain,
        )

    # Build risk description
    risk_descriptions = {
        "R0": "No clinical concern identified",
        "R1": "Low-level concern, routine follow-up",
        "R2": "Moderate concern, warm transfer to nurse",
        "R3": "Emergent — instruct patient to call 911 and transfer to nurse",
    }

    # Build domain nomination responses
    nom_responses = [
        DomainNominationResponse(
            domain=n.domain,
            confidence_tier=n.confidence_tier.value,
            priority=n.priority,
            target_risk_class=n.target_risk_class.value if n.target_risk_class else None,
            is_negated=n.is_negated,
            recommended_flow=n.recommended_flow,
            trigger_match_count=len(n.trigger_matches),
        )
        for n in nominations[:10]  # Cap at 10 for response size
    ]

    postprocess_time = (time.perf_counter() - postprocess_start) * 1000
    total_time = (time.perf_counter() - total_start) * 1000

    if ENABLE_METRICS:
        metrics.record_request(
            success=True,
            inference_ms=inference_time,
            tokenize_ms=tokenize_time,
            escalated=should_escalate,
        )

    # Audit log for clinical assessments
    if assessment.risk_class >= RiskClass.R2:
        audit_logger.warning(
            "CLINICAL_ESCALATION: risk=%s domain=%s hard_escalate=%s safety_override=%s",
            assessment.risk_class.value,
            assessment.primary_domain,
            assessment.hard_escalate,
            assessment.safety_override,
        )

    # Taxonomy scoring (deterministic)
    tax_scores: Dict[str, float] = {}
    matched_tax: List[TaxonomyMatchDetail] = []
    top_tax_domain: Optional[str] = None
    if TAXONOMY_SCORER_AVAILABLE and trigger_engine and decision_config:
        try:
            tax_result = score_taxonomy(predict_request.text, trigger_engine, decision_config)
            tax_scores = tax_result.scores
            top_tax_domain = tax_result.top_domain
            matched_tax = [
                TaxonomyMatchDetail(
                    domain=d.domain,
                    score=round(d.score, 4),
                    confidence_tier=d.confidence_tier,
                    priority=d.priority,
                    is_negated=d.is_negated,
                    target_risk_class=d.target_risk_class,
                    recommended_flow=d.recommended_flow,
                    trigger_count=d.trigger_count,
                    matched_phrases=d.matched_phrases,
                )
                for d in tax_result.matched_domains
            ]
        except Exception as e:
            logger.warning("Taxonomy scoring failed in /predict/clinical: %s", e)

    return ClinicalAssessmentResponse(
        predicted_label=final_label,
        probabilities=prob_dict,
        ml_confidence=ml_confidence,
        risk_class=assessment.risk_class.value,
        risk_description=risk_descriptions.get(
            assessment.risk_class.value, "Unknown"
        ),
        primary_domain=assessment.primary_domain,
        requires_911=assessment.requires_911,
        requires_nurse_transfer=assessment.requires_nurse_transfer,
        hard_escalate=assessment.hard_escalate,
        safety_override=assessment.safety_override,
        domain_nominations=nom_responses,
        taxonomy_scores=tax_scores,
        matched_taxonomy=matched_tax,
        top_taxonomy_domain=top_tax_domain,
        recommended_flow=assessment.recommended_flow,
        recommended_action=(
            assessment.recommended_action.value
            if assessment.recommended_action and hasattr(assessment.recommended_action, 'value')
            else assessment.recommended_action
        ),
        should_escalate=should_escalate,
        escalation_probability=escalation_prob,
        decision_reason=decision_reason,
        timing=TimingInfo(
            tokenize_ms=round(tokenize_time, 3),
            inference_ms=round(inference_time, 3),
            postprocess_ms=round(postprocess_time, 3),
            total_ms=round(total_time, 3),
        ),
    )


# =============================================================================
# PHASE 2: CONVERSATION FLOW ENDPOINTS
# =============================================================================

class ConversationStartRequest(BaseModel):
    """Request to start a new conversation session."""
    session_id: str = Field(..., min_length=1, max_length=128, description="Unique session identifier")
    patient_id: Optional[str] = Field(None, description="Patient identifier")
    tenant_id: Optional[str] = Field(None, description="Tenant identifier")


class ConversationTurnRequest(BaseModel):
    """Request to process a conversational turn."""
    session_id: str = Field(..., min_length=1, max_length=128)
    text: str = Field(..., min_length=1, max_length=MAX_INPUT_CHARS, description="Patient utterance")
    extracted_slots: Optional[Dict[str, Any]] = Field(None, description="Slots extracted by NLU")

    @field_validator("text", mode="before")
    @classmethod
    def sanitize_text(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return None
        v = bleach.clean(v, tags=[], strip=True)
        v = " ".join(v.split())
        v = v.replace("\x00", "").replace("\r", "")
        return v


class ResponseSpecResponse(BaseModel):
    """LLM response constraints for a flow step."""
    id: str = ""
    goal: str = ""
    tone: str = "professional"
    must_include: List[str] = Field(default_factory=list)
    must_ask: List[str] = Field(default_factory=list)
    must_not: List[str] = Field(default_factory=list)
    max_questions: int = 0


class ConversationTurnResponse(BaseModel):
    """Response from a conversation turn."""
    outcome: str = Field(..., description="proceed / clarify / escalate / handoff / end")
    flow_id: Optional[str] = None
    flow_type: Optional[str] = None
    step_number: int = 0
    step_type: Optional[str] = None
    response_spec: Optional[ResponseSpecResponse] = None
    prompt_constraints: Optional[str] = Field(None, description="Ready-to-use LLM prompt constraints")
    collects: List[str] = Field(default_factory=list)
    handoff_target: Optional[str] = None
    handoff_urgency: Optional[str] = None
    risk_class: Optional[str] = None
    domain: Optional[str] = None
    message: str = ""
    requires_reassessment: bool = False
    # Clinical assessment included when available
    clinical_assessment: Optional[Dict[str, Any]] = None
    timing: Optional[TimingInfo] = None


@app.post("/conversation/start", response_model=ConversationTurnResponse, tags=["Conversation"])
@limiter.limit(RATE_LIMIT)
async def conversation_start(
    request: Request,
    start_request: ConversationStartRequest,
    api_key: str = Depends(verify_api_key),
):
    """
    Start a new conversation session.

    Creates a session and returns the first flow step (greeting).
    The response_spec contains LLM constraints for generating Avery's response.
    """
    if not DECISION_ENGINE_AVAILABLE or flow_engine is None:
        raise HTTPException(status_code=503, detail="Flow engine not loaded")

    total_start = time.perf_counter()

    session_state = flow_engine.create_session(
        session_id=start_request.session_id,
        patient_id=start_request.patient_id,
        tenant_id=start_request.tenant_id,
    )
    result = flow_engine.start_call(session_state)

    total_time = (time.perf_counter() - total_start) * 1000

    return _turn_result_to_response(result, total_time)


@app.post("/conversation/turn", response_model=ConversationTurnResponse, tags=["Conversation"])
@limiter.limit(RATE_LIMIT)
async def conversation_turn(
    request: Request,
    turn_request: ConversationTurnRequest,
    api_key: str = Depends(verify_api_key),
):
    """
    Process a patient turn in an active conversation.

    Runs ML inference + taxonomy triggers + risk assessment, then advances
    the conversation flow. Returns the next response_spec for LLM generation.

    Safety behavior:
    - R3 emergencies interrupt any active flow immediately
    - Hard-escalate domains (suicidal_ideation, homicidal_ideation) always escalate
    - Response specs enforce must_not constraints (e.g., never say "heart attack")
    """
    if session is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    if not DECISION_ENGINE_AVAILABLE or flow_engine is None:
        raise HTTPException(status_code=503, detail="Flow engine not loaded")

    session_state = flow_engine.get_session(turn_request.session_id)
    if session_state is None:
        raise HTTPException(status_code=404, detail=f"Session '{turn_request.session_id}' not found")

    total_start = time.perf_counter()

    # Temporal preprocessor + safety keyword check (safety net for /conversation/turn)
    temporal_signal = None
    text_for_model = turn_request.text
    if TEMPORAL_PREPROCESSOR_AVAILABLE:
        temporal_signal = preprocess_temporal(turn_request.text)
        text_for_model = temporal_signal.processed_text

    # Step 1: ML inference on patient text
    tokenize_start = time.perf_counter()
    inputs = tokenizer(
        text_for_model,
        padding="max_length",
        truncation=True,
        max_length=MAX_LENGTH,
        return_tensors="np",
    )
    tokenize_time = (time.perf_counter() - tokenize_start) * 1000

    input_ids = inputs["input_ids"].astype(np.int64)
    attention_mask = inputs["attention_mask"].astype(np.int64)
    token_type_ids = None
    input_names_list = [inp.name for inp in session.get_inputs()]
    if "token_type_ids" in input_names_list:
        if "token_type_ids" in inputs:
            token_type_ids = inputs["token_type_ids"].astype(np.int64)
        else:
            token_type_ids = np.zeros_like(input_ids, dtype=np.int64)

    inference_start = time.perf_counter()
    logits = await run_inference_async(input_ids, attention_mask, token_type_ids)
    inference_time = (time.perf_counter() - inference_start) * 1000

    probabilities = softmax(logits)
    argmax_id = int(np.argmax(probabilities))
    argmax_label = id2label.get(argmax_id, "UNKNOWN")
    ml_confidence = float(np.max(probabilities))

    prob_dict = {
        id2label.get(i, f"LABEL_{i}"): float(p)
        for i, p in enumerate(probabilities)
    }

    # Step 2: Domain nomination + risk assessment
    postprocess_start = time.perf_counter()
    nominations = domain_nominator.nominate(
        text=turn_request.text,
        ml_label=argmax_label,
        ml_probabilities=prob_dict,
    )
    assessment = risk_classifier.assess(
        nominations=nominations,
        slot_values=dict(session_state.slots.slots),
        patient_context=None,
        ml_label=argmax_label,
        ml_confidence=ml_confidence,
    )

    # Safety keyword check (was missing from /conversation/turn — CRIT-13 fix)
    keyword_escalate, keyword_reason = check_safety_keywords(turn_request.text)
    if keyword_escalate and assessment.risk_class < RiskClass.R2:
        assessment = RiskAssessment(
            risk_class=RiskClass.R2,
            primary_domain=assessment.primary_domain,
            domain_nominations=assessment.domain_nominations,
            matched_rules=assessment.matched_rules,
            suppressions=assessment.suppressions,
            hard_escalate=assessment.hard_escalate,
            safety_override=True,
            ml_label=assessment.ml_label,
            ml_confidence=assessment.ml_confidence,
            recommended_flow="handoff",
            recommended_action=assessment.recommended_action,
        )

    # Temporal preprocessor override
    if temporal_signal and temporal_signal.is_active_emergency and assessment.risk_class < RiskClass.R2:
        assessment = RiskAssessment(
            risk_class=RiskClass.R2,
            primary_domain=assessment.primary_domain,
            domain_nominations=assessment.domain_nominations,
            matched_rules=assessment.matched_rules,
            suppressions=assessment.suppressions,
            hard_escalate=assessment.hard_escalate,
            safety_override=True,
            ml_label=assessment.ml_label,
            ml_confidence=assessment.ml_confidence,
            recommended_flow="handoff",
            recommended_action=assessment.recommended_action,
        )

    # CONTRASTIVE REVERSAL OVERRIDE (mirrors /predict/clinical logic):
    # When temporal BUT-clause reversal confirms active condition, override
    # domain negation for R3 domains and force R3 escalation.
    if (
        temporal_signal
        and temporal_signal.is_active_emergency
        and any(
            r.startswith("but_clause_reversal") or r == "contrastive_post_active"
            for r in temporal_signal.matched_rules
        )
        and assessment.risk_class < RiskClass.R3
    ):
        override_domain = None
        if decision_config:
            hard_esc_domains = decision_config.hard_escalate_domains
            for nom in nominations:
                if nom.domain in hard_esc_domains and nom.is_negated:
                    override_domain = nom.domain
                    break
        if override_domain:
            assessment = RiskAssessment(
                risk_class=RiskClass.R3,
                primary_domain=override_domain,
                domain_nominations=assessment.domain_nominations,
                matched_rules=assessment.matched_rules,
                suppressions=assessment.suppressions,
                hard_escalate=True,
                safety_override=True,
                ml_label=assessment.ml_label,
                ml_confidence=assessment.ml_confidence,
                recommended_flow="escalate_r3",
                recommended_action=None,
            )
            logger.warning(
                "Contrastive reversal override (/conversation/turn): "
                "%s negation overridden, temporal confirms active — R3",
                override_domain,
            )

    # Step 3: Process turn through flow engine
    result = flow_engine.process_turn(
        session=session_state,
        patient_text=turn_request.text,
        assessment=assessment,
        extracted_slots=turn_request.extracted_slots,
    )

    postprocess_time = (time.perf_counter() - postprocess_start) * 1000
    total_time = (time.perf_counter() - total_start) * 1000

    if ENABLE_METRICS:
        metrics.record_request(
            success=True,
            inference_ms=inference_time,
            tokenize_ms=tokenize_time,
            escalated=result.is_escalation,
        )

    if result.is_escalation:
        audit_logger.warning(
            "CONVERSATION_ESCALATION: session=%s domain=%s risk=%s",
            turn_request.session_id,
            result.domain,
            result.risk_class.value if result.risk_class else "unknown",
        )

    response = _turn_result_to_response(
        result, total_time, tokenize_time, inference_time, postprocess_time
    )
    response.clinical_assessment = assessment.to_dict()
    return response


@app.post("/conversation/end", tags=["Conversation"])
@limiter.limit(RATE_LIMIT)
async def conversation_end(
    request: Request,
    session_id: str = Query(..., min_length=1),
    api_key: str = Depends(verify_api_key),
):
    """End and clean up a conversation session."""
    if flow_engine is None:
        raise HTTPException(status_code=503, detail="Flow engine not loaded")
    flow_engine.destroy_session(session_id)
    return {"status": "ended", "session_id": session_id}


def _turn_result_to_response(
    result,
    total_ms: float,
    tokenize_ms: float = 0.0,
    inference_ms: float = 0.0,
    postprocess_ms: float = 0.0,
) -> ConversationTurnResponse:
    """Convert a TurnResult to a ConversationTurnResponse."""
    spec_response = None
    if result.response_spec:
        spec_response = ResponseSpecResponse(
            id=result.response_spec.spec_id,
            goal=result.response_spec.goal,
            tone=result.response_spec.tone,
            must_include=result.response_spec.must_include,
            must_ask=result.response_spec.must_ask,
            must_not=result.response_spec.must_not,
            max_questions=result.response_spec.max_questions,
        )

    return ConversationTurnResponse(
        outcome=result.outcome.value,
        flow_id=result.flow_id,
        flow_type=result.flow_type,
        step_number=result.step_number,
        step_type=result.step_type,
        response_spec=spec_response,
        prompt_constraints=result.prompt_constraints,
        collects=result.collects,
        handoff_target=result.handoff_target,
        handoff_urgency=result.handoff_urgency,
        risk_class=result.risk_class.value if result.risk_class else None,
        domain=result.domain,
        message=result.message,
        requires_reassessment=result.requires_reassessment,
        timing=TimingInfo(
            tokenize_ms=round(tokenize_ms, 3),
            inference_ms=round(inference_ms, 3),
            postprocess_ms=round(postprocess_ms, 3),
            total_ms=round(total_ms, 3),
        ),
    )


class MultiIntentResponse(BaseModel):
    """Response for multi-intent detection."""
    detected_intents: List[dict] = Field(..., description="List of detected intents with probabilities")
    primary_intent: str = Field(..., description="Highest probability intent")
    should_escalate: bool = Field(..., description="Whether escalation is needed")
    taxonomy_scores: Dict[str, float] = Field(default_factory=dict, description="Deterministic match scores for all taxonomy domains")
    matched_taxonomy: List[TaxonomyMatchDetail] = Field(default_factory=list, description="Taxonomy domains that matched with details")
    top_taxonomy_domain: Optional[str] = Field(None, description="Highest-scoring taxonomy domain")
    timing: TimingInfo


def split_sentences(text: str) -> List[str]:
    """
    Split text into sentences for multi-intent detection.
    
    Handles:
    - Standard punctuation (. ! ?)
    - Conjunctions (and, but, also, plus)
    - Semicolons and commas with conjunctions
    - "I also need", "I also want" patterns
    
    Args:
        text: Input text to split
        
    Returns:
        List of sentence fragments for individual classification
    """
    # First, split on standard sentence terminators
    parts = re.split(r'[.!?;]', text)
    
    # Further split on conjunctions that indicate separate intents
    expanded_parts = []
    conjunction_pattern = re.compile(
        r'\s+(?:but|and also|also|plus|additionally|as well as)\s+',
        re.IGNORECASE
    )
    # Pattern for "and I" which often indicates a new intent
    and_i_pattern = re.compile(r'\s+and\s+(?=I\s)', re.IGNORECASE)
    # Pattern for comma followed by "I need/want/have"
    comma_i_pattern = re.compile(r',\s*(?=I\s+(?:need|want|have|also))', re.IGNORECASE)
    
    for part in parts:
        # Split on conjunctions
        sub_parts = conjunction_pattern.split(part)
        for sub in sub_parts:
            # Split on "and I"
            sub_sub = and_i_pattern.split(sub)
            for s in sub_sub:
                # Split on ", I need/want"
                final_parts = comma_i_pattern.split(s)
                expanded_parts.extend(final_parts)
    
    # Clean and filter - minimum 5 chars to catch short phrases
    cleaned = [p.strip() for p in expanded_parts if p and len(p.strip()) >= 5]
    return cleaned if cleaned else [text]


@app.post("/predict/multi", response_model=MultiIntentResponse)
@limiter.limit(RATE_LIMIT)
async def predict_multi_intent(
    request: Request,
    predict_request: PredictRequest,
    threshold: float = Query(default=0.5, ge=0.0, le=1.0, description="Minimum probability threshold (0.0-1.0)"),
    api_key: str = Depends(verify_api_key),
):
    """
    Detect multiple intents by splitting compound sentences.
    
    Splits text on periods, 'but', 'and' conjunctions, then classifies each part.
    Returns unique intents detected across all parts.
    
    Args:
        threshold: Minimum probability to include an intent (default 0.5 = 50%, range 0.0-1.0)
    """
    # Validate threshold bounds
    if not 0.0 <= threshold <= 1.0:
        raise HTTPException(status_code=400, detail="threshold must be between 0.0 and 1.0")
    
    if session is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    total_start = time.perf_counter()
    
    text = predict_request.text
    context = predict_request.context
    
    # Temporal preprocessor
    temporal_signal = None
    if TEMPORAL_PREPROCESSOR_AVAILABLE:
        temporal_signal = preprocess_temporal(text, context)
        text = temporal_signal.processed_text
    
    # SAFETY FIRST: Check for critical keywords in original text only
    # (not context-concatenated text, to avoid false escalation from historical context)
    escalation_label = decision_policy.get("escalation_policy", {}).get("escalation_label", "ESCALATION")
    escalation_threshold = decision_policy.get("escalation_policy", {}).get("escalation_threshold", 0.5)
    
    keyword_escalate, keyword_reason = check_safety_keywords(predict_request.text)
    if keyword_escalate:
        # Immediate escalation - don't even bother with full analysis
        total_time = (time.perf_counter() - total_start) * 1000
        # Taxonomy scoring even on safety override
        tax_scores_multi: Dict[str, float] = {}
        matched_tax_multi: List[TaxonomyMatchDetail] = []
        top_tax_multi: Optional[str] = None
        if TAXONOMY_SCORER_AVAILABLE and trigger_engine and decision_config:
            try:
                tax_r = score_taxonomy(predict_request.text, trigger_engine, decision_config)
                tax_scores_multi = tax_r.scores
                top_tax_multi = tax_r.top_domain
                matched_tax_multi = [
                    TaxonomyMatchDetail(domain=d.domain, score=round(d.score, 4), confidence_tier=d.confidence_tier,
                        priority=d.priority, is_negated=d.is_negated, target_risk_class=d.target_risk_class,
                        recommended_flow=d.recommended_flow, trigger_count=d.trigger_count, matched_phrases=d.matched_phrases)
                    for d in tax_r.matched_domains
                ]
            except Exception:
                pass
        return MultiIntentResponse(
            detected_intents=[{"intent": escalation_label, "probability": 1.0, "reason": f"SAFETY_OVERRIDE: {keyword_reason}"}],
            primary_intent=escalation_label,
            should_escalate=True,
            taxonomy_scores=tax_scores_multi,
            matched_taxonomy=matched_tax_multi,
            top_taxonomy_domain=top_tax_multi,
            timing=TimingInfo(tokenize_ms=0, inference_ms=0, postprocess_ms=0, total_ms=round(total_time, 3)),
        )
    
    # Split into sentence parts (text only — context is passed separately to tokenizer)
    parts = split_sentences(text)
    
    detected_map = {}  # intent -> max probability
    should_escalate = False
    
    tokenize_time = 0
    inference_time = 0
    
    for part in parts:
        # Tokenize: text (part) as Segment A, context as Segment B (truncated first)
        t_start = time.perf_counter()
        if context:
            inputs = tokenizer(part, text_pair=context, padding="max_length", truncation="only_second", max_length=MAX_LENGTH, return_tensors="np")
        else:
            inputs = tokenizer(part, padding="max_length", truncation=True, max_length=MAX_LENGTH, return_tensors="np")
        tokenize_time += (time.perf_counter() - t_start) * 1000
        
        input_ids = inputs["input_ids"].astype(np.int64)
        attention_mask = inputs["attention_mask"].astype(np.int64)
        token_type_ids = None
        input_names = [inp.name for inp in session.get_inputs()]
        if "token_type_ids" in input_names:
            token_type_ids = inputs.get("token_type_ids", np.zeros_like(input_ids)).astype(np.int64)
        
        # Inference
        i_start = time.perf_counter()
        logits = await run_inference_async(input_ids, attention_mask, token_type_ids)
        inference_time += (time.perf_counter() - i_start) * 1000
        
        probabilities = softmax(logits)
        
        # Get top intent for this part
        top_idx = int(np.argmax(probabilities))
        top_label = id2label.get(top_idx, f"LABEL_{top_idx}")
        top_prob = float(probabilities[top_idx])
        
        if top_prob >= threshold:
            if top_label not in detected_map or detected_map[top_label] < top_prob:
                detected_map[top_label] = top_prob
        
        # Check escalation
        esc_prob = float(probabilities[label2id.get(escalation_label, 0)])
        if esc_prob >= escalation_threshold:
            should_escalate = True
    
    postprocess_start = time.perf_counter()
    
    # Build response
    detected = [{"intent": k, "probability": round(v, 4)} for k, v in detected_map.items()]
    detected.sort(key=lambda x: x["probability"], reverse=True)
    
    primary_intent = detected[0]["intent"] if detected else "UNKNOWN"
    
    # Temporal preprocessor override for multi-intent
    if temporal_signal and temporal_signal.is_active_emergency and not should_escalate:
        should_escalate = True
        if escalation_label not in detected_map:
            detected.insert(0, {"intent": escalation_label, "probability": 1.0, "reason": f"TEMPORAL_OVERRIDE: {temporal_signal.reason}"})
        primary_intent = escalation_label
    
    postprocess_time = (time.perf_counter() - postprocess_start) * 1000
    total_time = (time.perf_counter() - total_start) * 1000
    
    # Taxonomy scoring (deterministic)
    tax_scores_m: Dict[str, float] = {}
    matched_tax_m: List[TaxonomyMatchDetail] = []
    top_tax_m: Optional[str] = None
    if TAXONOMY_SCORER_AVAILABLE and trigger_engine and decision_config:
        try:
            tax_r = score_taxonomy(predict_request.text, trigger_engine, decision_config)
            tax_scores_m = tax_r.scores
            top_tax_m = tax_r.top_domain
            matched_tax_m = [
                TaxonomyMatchDetail(domain=d.domain, score=round(d.score, 4), confidence_tier=d.confidence_tier,
                    priority=d.priority, is_negated=d.is_negated, target_risk_class=d.target_risk_class,
                    recommended_flow=d.recommended_flow, trigger_count=d.trigger_count, matched_phrases=d.matched_phrases)
                for d in tax_r.matched_domains
            ]
            # SAFETY NET: hard-escalate R3 domains override ML
            hard_esc_domains = decision_config.hard_escalate_domains
            for d in tax_r.matched_domains:
                if d.domain in hard_esc_domains and not d.is_negated and d.score > 0:
                    if not should_escalate:
                        should_escalate = True
                        primary_intent = escalation_label
                        if escalation_label not in detected_map:
                            detected.insert(0, {"intent": escalation_label, "probability": 1.0, "reason": f"TAXONOMY_ESCALATION: {d.domain}"})
                    break
        except Exception as e:
            logger.warning("Taxonomy scoring failed in /predict/multi: %s", e)

    return MultiIntentResponse(
        detected_intents=detected,
        primary_intent=primary_intent,
        should_escalate=should_escalate,
        taxonomy_scores=tax_scores_m,
        matched_taxonomy=matched_tax_m,
        top_taxonomy_domain=top_tax_m,
        timing=TimingInfo(
            tokenize_ms=round(tokenize_time, 3),
            inference_ms=round(inference_time, 3),
            postprocess_ms=round(postprocess_time, 3),
            total_ms=round(total_time, 3),
        ),
    )


@app.get("/taxonomy", tags=["Discovery"])
@limiter.limit("30/minute")
async def get_taxonomy(
    request: Request,
    api_key: str = Depends(verify_api_key),
):
    """
    List all available taxonomy domains with their priority and target risk class.

    Returns every domain the deterministic engine can match against,
    along with metadata from each domain's rules.yaml.
    """
    if not decision_config:
        return {"domains": [], "count": 0, "message": "Decision engine not loaded"}

    domains = []
    for name in sorted(decision_config.taxonomy_triggers.keys()):
        priority = decision_config.get_domain_priority(name)
        target_risk = decision_config.get_domain_target_risk(name)
        domains.append({
            "domain": name,
            "display_name": name.replace("_", " ").title(),
            "priority": priority,
            "target_risk_class": target_risk,
        })
    return {"domains": domains, "count": len(domains)}


@app.get("/labels")
@limiter.limit("30/minute")
async def get_labels(
    request: Request,
    api_key: str = Depends(verify_api_key),
):
    """Get available intent labels (requires authentication)."""
    return {"labels": list(label2id.keys())}


@app.get("/policy")
@limiter.limit("10/minute")
async def get_policy(
    request: Request,
    api_key: str = Depends(verify_api_key),
):
    """
    Get current decision policy (requires authentication).
    
    This endpoint exposes sensitive threshold information.
    """
    audit_logger.info("Policy endpoint accessed")
    return decision_policy


@app.get("/metrics")
@limiter.limit("60/minute")
async def get_metrics(
    request: Request,
    api_key: str = Depends(verify_api_key),
):
    """
    Get service metrics for monitoring and observability.
    
    Returns request counts, latency averages, and throughput stats.
    Requires authentication.
    """
    if not ENABLE_METRICS:
        raise HTTPException(status_code=404, detail="Metrics disabled")
    
    return {
        "metrics": metrics.get_stats(),
        "config": {
            "inference_workers": INFERENCE_WORKERS,
            "onnx_intra_op_threads": ONNX_INTRA_OP_THREADS,
            "onnx_inter_op_threads": ONNX_INTER_OP_THREADS,
            "max_length": MAX_LENGTH,
            "batching_enabled": ENABLE_BATCHING,
        },
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


@app.get("/ready")
async def readiness_check(request: Request):
    """
    Kubernetes-style readiness probe.
    
    Returns 200 if service is ready to accept traffic,
    503 if not ready (model not loaded).
    No authentication required for orchestrator probes.
    """
    if session is None or tokenizer is None:
        raise HTTPException(status_code=503, detail="Service not ready")
    return {"status": "ready"}


@app.get("/compliance")
@limiter.limit("10/minute")
async def compliance_status(
    request: Request,
    api_key: str = Depends(verify_api_key),
):
    """
    Get HIPAA compliance configuration status.
    
    Returns current security settings and compliance controls.
    Requires authentication. For clinical deployment audits.
    """
    if not HIPAA_ENABLED:
        return {
            "hipaa_module_loaded": False,
            "message": "HIPAA compliance module not available",
        }
    
    return {
        "hipaa_module_loaded": True,
        "compliance_status": get_compliance_status(),
        "service_info": {
            "version": "1.0.0",
            "environment": os.getenv("ENVIRONMENT", "development"),
            "docs_disabled": DISABLE_DOCS,
            "auth_required": REQUIRE_AUTH,
        },
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }


# =============================================================================
# STREAMING INTENT ENDPOINTS
# =============================================================================

# Thread-safe session-based streaming tracker storage
_streaming_config = None
_streaming_config_lock = threading.Lock()
# Store tuple of (tracker, last_access_time) for TTL-based expiration
_session_trackers: OrderedDict[str, tuple] = OrderedDict()
_session_trackers_lock = threading.Lock()
MAX_SESSIONS = int(os.getenv("MAX_STREAMING_SESSIONS", "1000"))
SESSION_TTL_SECONDS = float(os.getenv("SESSION_TTL_SECONDS", "1800"))  # 30 minutes default


def _get_streaming_config():
    """Get or load streaming config (thread-safe singleton)."""
    global _streaming_config
    if _streaming_config is None:
        with _streaming_config_lock:
            if _streaming_config is None:
                from streaming_intent import StreamingConfig
                _streaming_config = StreamingConfig.from_yaml()
    return _streaming_config


def _run_streaming_inference_sync(text: str, config) -> Dict[str, float]:
    """Synchronous ONNX inference for streaming (runs in executor)."""
    if tokenizer is None or session is None:
        raise RuntimeError("Model not loaded")
    
    inputs = tokenizer(
        text,
        return_tensors="np",
        max_length=config.model_max_length,
        truncation=True,
        padding="max_length",
    )
    
    # Build input dict with available inputs
    input_dict = {
        "input_ids": inputs["input_ids"],
        "attention_mask": inputs["attention_mask"],
    }
    # Add token_type_ids only if model expects it
    input_names = [inp.name for inp in session.get_inputs()]
    if "token_type_ids" in input_names:
        if "token_type_ids" in inputs:
            input_dict["token_type_ids"] = inputs["token_type_ids"]
        else:
            input_dict["token_type_ids"] = np.zeros_like(inputs["input_ids"])
    
    outputs = session.run(None, input_dict)
    
    logits = outputs[0][0]
    exp_logits = np.exp(logits - np.max(logits))
    probs = exp_logits / exp_logits.sum()
    
    return {id2label[i]: float(probs[i]) for i in range(len(probs))}


async def _run_streaming_inference_async(text: str) -> Dict[str, float]:
    """Async wrapper for streaming inference using ThreadPoolExecutor."""
    config = _get_streaming_config()
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        inference_executor,
        _run_streaming_inference_sync,
        text,
        config,
    )


def _create_model_inference_fn() -> callable:
    """Create model inference function for streaming tracker (sync version for tracker API)."""
    def model_inference(text: str) -> Dict[str, float]:
        """Use loaded ONNX model for inference."""
        config = _get_streaming_config()
        return _run_streaming_inference_sync(text, config)
    
    return model_inference


def get_streaming_tracker(session_id: str):
    """
    Get or create streaming intent tracker for a specific session (thread-safe).
    
    Each session gets its own tracker to prevent state corruption between
    concurrent requests. Old sessions are evicted based on TTL or when MAX_SESSIONS is reached.
    """
    global _session_trackers
    current_time = time.time()
    
    with _session_trackers_lock:
        # First, clean up expired sessions (TTL-based expiration)
        expired_sessions = [
            sid for sid, (_, last_access) in _session_trackers.items()
            if current_time - last_access > SESSION_TTL_SECONDS
        ]
        for sid in expired_sessions:
            del _session_trackers[sid]
            logger.debug(f"Expired streaming session {sid}")
        
        if session_id in _session_trackers:
            # Update last access time and move to end (LRU)
            tracker, _ = _session_trackers[session_id]
            _session_trackers[session_id] = (tracker, current_time)
            _session_trackers.move_to_end(session_id)
            return tracker
        
        # Create new tracker for this session
        try:
            from streaming_intent import StreamingIntentTracker
            config = _get_streaming_config()
            
            tracker = StreamingIntentTracker(
                config=config,
                model_fn=_create_model_inference_fn(),
            )
            
            # Evict oldest session if at capacity
            while len(_session_trackers) >= MAX_SESSIONS:
                _session_trackers.popitem(last=False)
            
            _session_trackers[session_id] = (tracker, current_time)
            logger.debug(f"Created streaming tracker for session {session_id}")
            return tracker
            
        except ImportError as e:
            logger.warning(f"Streaming intent module not available: {e}")
            raise HTTPException(status_code=501, detail="Streaming intent module not installed")


def reset_streaming_tracker(session_id: str) -> bool:
    """Reset or remove a streaming tracker for a session."""
    with _session_trackers_lock:
        if session_id in _session_trackers:
            tracker, _ = _session_trackers[session_id]
            tracker.reset()
            _session_trackers[session_id] = (tracker, time.time())
            return True
        return False


class StreamChunkRequest(BaseModel):
    """Request schema for streaming chunk endpoint."""
    text: str = Field(..., min_length=1, max_length=MAX_INPUT_CHARS, description="STT chunk text")
    is_final: bool = Field(False, description="Whether this is a finalized STT segment")
    timestamp_ms: Optional[int] = Field(None, description="Optional timestamp in milliseconds")
    session_id: Optional[str] = Field(None, description="Session ID for tracking (auto-generated if not provided)")


class StreamDecisionResponse(BaseModel):
    """Response schema for streaming decision."""
    current_belief: Dict[str, float] = Field(..., description="Current probability distribution over intents")
    top_intent: str = Field(..., description="Intent with highest probability")
    top_intent_prob: float = Field(..., description="Probability of top intent")
    should_escalate: bool = Field(..., description="Whether escalation has been triggered")
    should_commit: bool = Field(..., description="Whether intent is ready to be committed")
    committed_intent: Optional[str] = Field(None, description="Final committed intent (if should_commit)")
    escalation_reason: Optional[str] = Field(None, description="Reason for escalation (if triggered)")
    window_text: str = Field(..., description="Current rolling window text")
    step_count: int = Field(..., description="Number of update steps processed")
    step_latency_ms: float = Field(..., description="Processing time for this step")
    # Decision engine enrichment (Phase 1 integration)
    risk_class: Optional[str] = Field(None, description="R0/R1/R2/R3 clinical risk tier")
    primary_domain: Optional[str] = Field(None, description="Primary clinical domain detected")
    domain_nominations: Optional[List[Dict[str, Any]]] = Field(None, description="Top domain nominations")


@app.post("/stream/chunk", response_model=StreamDecisionResponse, tags=["Streaming"])
@limiter.limit(RATE_LIMIT)
async def process_stream_chunk(
    request: Request,
    chunk: StreamChunkRequest,
    api_key: str = Depends(verify_api_key),
):
    """
    Process a streaming STT chunk and return belief state + decision.
    
    This endpoint implements HMM-style belief updating for stable intent
    classification over streaming text. Use for real-time STT processing.
    
    **Decision Rules:**
    - Immediate escalation if P(ESCALATION) >= 0.85
    - Escalate if P(ESCALATION) >= 0.60 for 3 consecutive steps
    - Commit non-escalation intent if top_prob >= 0.70 AND stable for 3 steps
    
    **Usage:**
    1. Send chunks as they arrive from STT
    2. Set `is_final=true` for finalized STT segments
    3. Check `should_escalate` for immediate clinical escalation
    4. Check `should_commit` and `committed_intent` for final classification
    """
    try:
        from streaming_intent import Chunk
        
        # Get or generate session ID for thread-safe tracking
        session_id = chunk.session_id or str(uuid.uuid4())
        tracker = get_streaming_tracker(session_id)
        
        chunk_obj = Chunk(
            text=chunk.text,
            is_final=chunk.is_final,
            timestamp_ms=chunk.timestamp_ms,
        )
        
        # Run tracker update in executor to prevent blocking the event loop
        # (tracker.update calls synchronous ONNX inference internally)
        loop = asyncio.get_running_loop()
        decision = await loop.run_in_executor(
            inference_executor, tracker.update, chunk_obj
        )
        
        # Safety override — use decision engine if available, else legacy keywords
        should_escalate = decision.should_escalate
        escalation_reason = decision.escalation_reason
        risk_class_str = None
        primary_domain_str = None
        domain_nom_list = None

        # Determine text to analyze (prefer window_text for cross-chunk safety)
        safety_text = decision.window_text or chunk.text

        if DECISION_ENGINE_AVAILABLE and domain_nominator is not None:
            # Always run decision engine for enrichment (risk_class, domain, nominations)
            nominations = domain_nominator.nominate(
                text=safety_text,
                ml_label=decision.top_intent,
                ml_probabilities=decision.current_belief,
            )
            assessment = risk_classifier.assess(
                nominations=nominations,
                ml_label=decision.top_intent,
                ml_confidence=decision.top_intent_prob,
            )
            risk_class_str = assessment.risk_class.value
            primary_domain_str = assessment.primary_domain

            # Upgrade escalation if engine detects R2+ and tracker hasn't already
            if not should_escalate and assessment.risk_class >= RiskClass.R2:
                should_escalate = True
                if assessment.hard_escalate:
                    escalation_reason = f"HARD_ESCALATE: {assessment.primary_domain} → {assessment.risk_class.value}"
                elif assessment.safety_override:
                    escalation_reason = f"SAFETY_OVERRIDE: {assessment.primary_domain} → {assessment.risk_class.value}"
                else:
                    escalation_reason = f"CLINICAL_RISK: {assessment.primary_domain} → {assessment.risk_class.value}"
            elif should_escalate and assessment.risk_class < RiskClass.R2:
                # Tracker escalated but engine says R0/R1 — belt+suspenders upgrade
                risk_class_str = "R2"

            domain_nom_list = [
                {"domain": n.domain, "confidence": n.confidence_tier.value, "negated": n.is_negated}
                for n in nominations[:5]
            ]
        elif not should_escalate:
            # Fallback to legacy safety keywords
            keyword_match, keyword_reason = check_safety_keywords(chunk.text)
            if not keyword_match and decision.window_text:
                keyword_match, keyword_reason = check_safety_keywords(decision.window_text)
            if keyword_match:
                should_escalate = True
                escalation_reason = keyword_reason
                risk_class_str = "R2"

        # Audit log escalations
        if should_escalate and HIPAA_ENABLED:
            hipaa_audit.log_escalation(
                request_id=getattr(request.state, 'request_id', 'unknown'),
                client_ip=request.client.host if request.client else 'unknown',
                escalation_type=escalation_reason or "streaming_escalation",
                confidence=decision.current_belief.get("ESCALATION", 0),
            )
        
        return StreamDecisionResponse(
            current_belief=decision.current_belief,
            top_intent=decision.top_intent,
            top_intent_prob=decision.top_intent_prob,
            should_escalate=should_escalate,
            should_commit=decision.should_commit,
            committed_intent=decision.committed_intent,
            escalation_reason=escalation_reason,
            window_text=decision.window_text,
            step_count=decision.step_count,
            step_latency_ms=round(decision.step_latency_ms, 3),
            risk_class=risk_class_str,
            primary_domain=primary_domain_str,
            domain_nominations=domain_nom_list,
        )
    except ImportError:
        raise HTTPException(status_code=501, detail="Streaming intent module not available")
    except Exception as e:
        logger.error(f"Streaming chunk error: {e}")
        raise HTTPException(status_code=500, detail="Streaming processing failed")


class StreamResetRequest(BaseModel):
    """Request schema for stream reset endpoint."""
    session_id: str = Field(..., min_length=1, description="Session ID to reset")


@app.post("/stream/reset", tags=["Streaming"])
@limiter.limit("30/minute")
async def reset_stream(
    request: Request,
    reset_request: StreamResetRequest,
    api_key: str = Depends(verify_api_key),
):
    """
    Reset streaming tracker state for a specific session.
    
    Call this when starting a new conversation/session to clear
    accumulated text buffer and belief state.
    """
    try:
        was_reset = reset_streaming_tracker(reset_request.session_id)
        if was_reset:
            return {"status": "reset", "session_id": reset_request.session_id, "message": "Streaming tracker reset"}
        else:
            return {"status": "not_found", "session_id": reset_request.session_id, "message": "Session not found (may already be expired)"}
    except ImportError:
        raise HTTPException(status_code=501, detail="Streaming intent module not available")


@app.get("/stream/metrics", tags=["Streaming"])
@limiter.limit("30/minute")
async def get_stream_metrics(
    request: Request,
    session_id: Optional[str] = None,
    api_key: str = Depends(verify_api_key),
):
    """
    Get streaming tracker metrics for a specific session or aggregate.
    
    Returns escalation count, gray zone rate, average steps to decision.
    """
    try:
        if session_id:
            with _session_trackers_lock:
                if session_id in _session_trackers:
                    tracker, _ = _session_trackers[session_id]
                    metrics_data = tracker.get_metrics()
                else:
                    raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
        else:
            # Return aggregate info about all sessions
            with _session_trackers_lock:
                metrics_data = {
                    "active_sessions": len(_session_trackers),
                    "max_sessions": MAX_SESSIONS,
                }
        return {
            "streaming_metrics": metrics_data,
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
    except ImportError:
        raise HTTPException(status_code=501, detail="Streaming intent module not available")


@app.get("/stream/config", tags=["Streaming"])
@limiter.limit("10/minute")
async def get_stream_config(
    request: Request,
    api_key: str = Depends(verify_api_key),
):
    """
    Get streaming intent configuration.
    
    Returns thresholds, transition matrix info, and window settings.
    """
    try:
        config = _get_streaming_config()
        return {
            "thresholds": {
                "theta_hi": config.theta_hi,
                "theta_med": config.theta_med,
                "theta_lock": config.theta_lock,
                "K": config.K,
            },
            "emission": {
                "alpha": config.alpha,
                "epsilon": config.epsilon,
            },
            "window": {
                "max_tokens": config.max_tokens,
                "max_tokens_limit": config.max_tokens_limit,
            },
            "debounce": {
                "debounce_ms": config.debounce_ms,
                "min_change_chars": config.min_change_chars,
            },
            "intents": config.intents,
        }
    except ImportError:
        raise HTTPException(status_code=501, detail="Streaming intent module not available")


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Global exception handler - never expose internal details."""
    request_id = getattr(request.state, 'request_id', 'unknown')
    # Log full error internally
    logger.error(f"Unhandled exception [request_id={request_id}]: {exc}", exc_info=True)
    # Return generic error to client
    return JSONResponse(
        status_code=500,
        content={
            "detail": "Internal server error",
            "request_id": request_id,  # Allow user to reference for support
        },
        headers={"X-Request-ID": request_id},
    )


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)