Recipe 11.2: Python Implementation Example

pip install boto3

import json
import logging
import re
import uuid
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional

import boto3
from botocore.config import Config

# Structured logging. In production, ship JSON-formatted records to
# CloudWatch Logs Insights for cross-call investigation. Conversation
# logs are PHI by association: the user's question may name visit
# reasons, providers, or other clinical context. Log structural
# metadata only (session_id, intent, tool name, tool latency,
# tool outcome, identity-verification path), never raw user
# utterances, never raw generated responses, never tool arguments
# that contain identifiers, never the patient name or DOB collected
# during identity verification. The full transcripts and full tool
# calls live in the audit pipeline (Firehose + Object-Lock S3) with
# appropriate access controls.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from Bedrock, DynamoDB,
# EventBridge, Firehose, CloudWatch, S3, and Secrets Manager. The
# scheduling-bot response window is tight: the patient is staring
# at the chat widget waiting, and a retry storm that adds 5 seconds
# of latency is operationally worse than a fast failure with a
# graceful degraded-mode message. Cap the retries and let the
# graceful-failure path handle the fall-back.
BOTO3_RETRY_CONFIG = Config(
    retries={"max_attempts": 3, "mode": "adaptive"})

# Module-level clients. Reused across Lambda invocations in warm
# containers so each invocation does not pay the connection cost.
REGION = "us-east-1"
bedrock_runtime       = boto3.client("bedrock-runtime",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
bedrock_agent_runtime = boto3.client("bedrock-agent-runtime",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
dynamodb              = boto3.resource("dynamodb",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
eventbridge_client    = boto3.client("events",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
firehose_client       = boto3.client("firehose",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
cloudwatch_client     = boto3.client("cloudwatch",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
s3_client             = boto3.client("s3",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
secrets_client        = boto3.client("secretsmanager",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Fill these in with your actual resource names. The demo prints
# what it would write rather than failing if the resources do
# not exist; see run_demo() at the bottom.
CONVERSATION_STATE_TABLE      = "scheduling-bot-conversation-state"
CONVERSATION_METADATA_TABLE   = "scheduling-bot-conversation-metadata"
TOOL_CALL_LEDGER_TABLE        = "scheduling-bot-tool-call-ledger"
SCHEDULING_EVENT_BUS_NAME     = "scheduling-bot-events-bus"
AUDIT_ARCHIVE_FIREHOSE_NAME   = "scheduling-bot-audit-archive"
BOOKING_EVENT_JOURNAL_BUCKET  = "scheduling-bot-booking-journal"
CLOUDWATCH_NAMESPACE          = "SchedulingBot"

# Bedrock Knowledge Base ID for visit-type catalog and provider
# content. The Knowledge Base owns chunking, embedding, and the
# vector store.
KNOWLEDGE_BASE_ID             = "KB_PLACEHOLDER_ID"

# Bedrock Guardrail for restricted-topic filtering. Configure in
# the Bedrock console with restricted topics for clinical-advice
# and account-specific-billing at minimum. Pin to a specific
# version, not DRAFT, in production.
GUARDRAIL_ID                  = "GUARDRAIL_PLACEHOLDER_ID"
GUARDRAIL_VERSION             = "1"

# Deploy-time guardrail. Any blank resource name is a deploy-time
# bug, not a runtime surprise.
for _name, _value in [
    ("CONVERSATION_STATE_TABLE",     CONVERSATION_STATE_TABLE),
    ("CONVERSATION_METADATA_TABLE",  CONVERSATION_METADATA_TABLE),
    ("TOOL_CALL_LEDGER_TABLE",       TOOL_CALL_LEDGER_TABLE),
    ("SCHEDULING_EVENT_BUS_NAME",    SCHEDULING_EVENT_BUS_NAME),
    ("AUDIT_ARCHIVE_FIREHOSE_NAME",  AUDIT_ARCHIVE_FIREHOSE_NAME),
    ("BOOKING_EVENT_JOURNAL_BUCKET", BOOKING_EVENT_JOURNAL_BUCKET),
    ("CLOUDWATCH_NAMESPACE",         CLOUDWATCH_NAMESPACE),
    ("KNOWLEDGE_BASE_ID",            KNOWLEDGE_BASE_ID),
    ("GUARDRAIL_ID",                 GUARDRAIL_ID),
    ("GUARDRAIL_VERSION",            GUARDRAIL_VERSION),
]:
    assert _value, f"{_name} must be set before deploying."

# --- Versioning ---
# Every conversation turn carries the model version, the prompt
# version, the Knowledge Base version, the Guardrail version, the
# visit-type catalog version, and the scheduling-policy version
# active at the time of the turn. This is how a future audit
# reconstructs which calibration was active when a particular
# booking happened.
PROMPT_VERSION                = "scheduling-bot-prompt-v2.4"
AGENT_VERSION                 = "scheduling-agent-v3.1"
VISIT_TYPE_CATALOG_VERSION    = "vt-catalog-2026-04-15"
SCHEDULING_POLICY_VERSION     = "policy-2026-03-01"
INSTITUTION_ID                = "riverside-clinic"
INSTITUTION_DISPLAY_NAME      = "Riverside Clinic"

# --- Model IDs ---
# Two model roles. Intent classification and parameter extraction
# are cheap per-turn tasks where a smaller model earns its keep.
# The orchestration model decides which tool to call when, with
# what arguments, given the patient's request and the recent
# conversation history; that work benefits from a stronger model.
#
# If your region requires cross-region inference, use the inference
# profile ID (e.g., "us.anthropic.claude-3-5-haiku-20241022-v1:0").
# TODO: verify the exact model IDs available in your region and
# account; Bedrock model availability evolves over time.
INTENT_CLASSIFIER_MODEL_ID    = "anthropic.claude-3-5-haiku-20241022-v1:0"
ORCHESTRATION_MODEL_ID        = "anthropic.claude-3-5-sonnet-20241022-v2:0"

# --- Pipeline Tuning ---
# Below this confidence, we ask a clarifying question rather than
# acting on the classification. Keep this on the conservative side;
# better to ask than to misroute.
INTENT_CONFIDENCE_THRESHOLD   = Decimal("0.65")

# Below this confidence on visit-type mapping, ask the patient to
# clarify rather than booking the wrong visit type.
VISIT_TYPE_CONFIDENCE_THRESHOLD = Decimal("0.7")

# Slot-hold TTL. Long enough that a patient deliberating over options
# has time to choose, short enough that abandoned holds release back
# into the inventory quickly.
HOLD_TTL_SECONDS              = 300

# --- Intents ---
SCHEDULING_INTENTS = [
    "new_appointment",
    "reschedule_appointment",
    "cancel_appointment",
    "check_appointment",
    "update_preferences",
]

OUT_OF_SCOPE_INTENTS = {
    "clinical_question":   "nurse_triage",
    "refill_request":      "refill_bot",
    "benefits_eligibility": "benefits_navigator",
    "general_question":    "faq_bot",
    "out_of_scope":        "live_agent",
}

# --- Identity Verification Policy ---
# Maps (intent, authenticated_session, hours_to_appointment) to a
# required assurance level. The privacy officer owns this policy
# in production; this Python dict is a placeholder for what is
# normally a versioned governance artifact stored in Parameter
# Store or a dedicated policy service.
#
# Assurance levels:
#   "authenticated": patient logged in via portal; trust the
#                    session-bound patient_id directly.
#   "basic":         name + DOB + one confirmation factor (last 4
#                    of phone, ZIP code, or one-time code).
#   "step_up":       basic plus a one-time code to a verified
#                    channel; used for higher-risk actions.
IDENTITY_POLICY = {
    # Intent: list of (predicate_lambda, required_level) rules,
    # evaluated in order; first matching rule wins.
    "new_appointment": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "basic"),
    ],
    "reschedule_appointment": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx:
            ctx.get("hours_to_appointment", 999) < 24,
         "step_up"),
        (lambda ctx: True, "basic"),
    ],
    "cancel_appointment": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx:
            ctx.get("hours_to_appointment", 999) < 24,
         "step_up"),
        (lambda ctx: True, "basic"),
    ],
    "check_appointment": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "basic"),
    ],
    "update_preferences": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "basic"),
    ],
}

# Confidence thresholds for the patient_lookup tool. The lookup
# returns a match score; below the threshold for the assurance
# level we either step up or hand off.
ASSURANCE_MATCH_THRESHOLDS = {
    "authenticated": Decimal("0.0"),  # already authenticated
    "basic":         Decimal("0.85"),
    "step_up":       Decimal("0.95"),
}

# --- Visit-Type Catalog (illustrative) ---
# In production this lives in a Bedrock Knowledge Base with each
# visit type as its own document chunk: name, purpose, duration,
# scheduling rules, prep instructions, eligible providers,
# accepted insurance plans. The catalog cleanup is the single
# highest-leverage operational investment for the bot's quality.
# The dict below is a starter for the demo only.
VISIT_TYPE_CATALOG = {
    "cardiology_established_followup_30min": {
        "display_name":
            "Cardiology Established Patient Follow-Up",
        "duration_minutes": 30,
        "description":
            "Follow-up visit with an established cardiologist "
            "after a procedure, test, or prior visit.",
        "natural_language_cues": [
            "follow up with my cardiologist",
            "follow up after stress test",
            "follow up after echo",
            "post-procedure cardiology",
        ],
        "specialty":   "cardiology",
        "patient_class": "established",
        "self_schedulable": True,
    },
    "cardiology_new_patient_60min": {
        "display_name":
            "Cardiology New Patient Consultation",
        "duration_minutes": 60,
        "description":
            "Initial cardiology consultation for a new "
            "patient referred for evaluation.",
        "natural_language_cues": [
            "first time seeing a cardiologist",
            "new patient cardiology",
            "see a cardiologist for the first time",
        ],
        "specialty":   "cardiology",
        "patient_class": "new",
        # New cardiology patients are not self-schedulable;
        # they are triaged first.
        "self_schedulable": False,
    },
    "primary_care_followup_20min": {
        "display_name":
            "Primary Care Follow-Up",
        "duration_minutes": 20,
        "description":
            "Routine follow-up with an established primary "
            "care provider.",
        "natural_language_cues": [
            "follow up with my doctor",
            "primary care follow-up",
            "see my pcp",
        ],
        "specialty":   "primary_care",
        "patient_class": "established",
        "self_schedulable": True,
    },
    "primary_care_annual_wellness_45min": {
        "display_name":
            "Annual Wellness Visit",
        "duration_minutes": 45,
        "description":
            "Yearly preventive wellness visit with a primary "
            "care provider.",
        "natural_language_cues": [
            "annual checkup",
            "yearly physical",
            "wellness visit",
            "annual physical",
        ],
        "specialty":   "primary_care",
        "patient_class": "any",
        "self_schedulable": True,
    },
}

# --- High-Acuity Clinical Cues ---
# When detected in the patient's reason-for-visit, route to the
# triage path rather than self-scheduling. Production uses
# Comprehend Medical's clinical-entity detection plus a tuned
# acuity classifier; the keyword backstop below is illustrative.
HIGH_ACUITY_CUES = [
    "chest pain",
    "chest pressure",
    "shortness of breath",
    "trouble breathing",
    "heart attack",
    "stroke",
    "severe headache",
    "worst headache of my life",
    "uncontrolled bleeding",
    "passed out",
    "loss of consciousness",
    "suicidal",
    "want to hurt myself",
    "want to end my life",
]

# --- Refusal-and-Handoff Templates ---
OUT_OF_SCOPE_HANDOFFS = {
    "clinical_question": (
        "That sounds like a question for our nurse advice "
        "line. I'm not a clinician and can't give medical "
        "advice. You can reach our nurse line at 555-0100, "
        "or in an emergency please call 911."
    ),
    "refill_request": (
        "Refill requests go through our pharmacy team. The "
        "fastest path is the patient portal's prescription "
        "section, or you can call our pharmacy line at "
        "555-0175."
    ),
    "benefits_eligibility": (
        "Plan-specific coverage questions need our benefits "
        "team. They can look up your account and answer "
        "specifically. Reach them at 555-0145."
    ),
    "general_question": (
        "That sounds like a general clinic question. Let me "
        "connect you with our clinic info assistant for that."
    ),
    "out_of_scope": (
        "Let me get you to a person who can help with that. "
        "Our front desk is at 555-0100."
    ),
}

GREETING_AND_DISCLOSURE_TEMPLATE = (
    f"Hi! I'm {INSTITUTION_DISPLAY_NAME}'s scheduling "
    "assistant. I can help you book, reschedule, or cancel "
    "appointments. I can't help with clinical questions or "
    "your account-specific billing, but I can connect you "
    "with the right team for those. How can I help today?"
)

CRISIS_ROUTE_TEMPLATE = (
    "If this is a medical emergency, please call 911 right "
    "now. For urgent symptoms that aren't a 911 emergency, "
    "our nurse advice line at 555-0100 can help right away. "
    "I'm a chatbot, so I can't book the right kind of "
    "appointment for an urgent situation. Let me connect "
    "you with a person who can."
)

NO_SLOTS_TEMPLATE = (
    "I'm not finding any open slots that match what you "
    "asked for. A few options: I can put you on the "
    "wait-list for a cancellation, look at a different "
    "provider or location, check telehealth availability, "
    "or connect you with our scheduling team. Which would "
    "you like?"
)

SLOT_RACE_TEMPLATE = (
    "Looks like that slot was taken by someone else just "
    "now. Let me grab fresh options for you."
)

HOLD_FAILED_TEMPLATE = (
    "I'm having trouble holding that slot. Let me get you "
    "to our scheduling team so we don't lose your booking "
    "to a system hiccup."
)

SYSTEM_ERROR_TEMPLATE = (
    "I'm having trouble reaching our scheduling system "
    "right now. I've made a note for our scheduling team "
    "to follow up with you, or you can call us at "
    "555-0150 to book directly."
)

GENERIC_FAILURE_TEMPLATE = (
    "Something went wrong on my end. Let me get you to "
    "our scheduling team at 555-0150 so this doesn't "
    "fall through the cracks."
)

NO_APPOINTMENT_FOUND_TEMPLATE = (
    "I'm not finding an upcoming appointment for you that "
    "matches what you described. Could you tell me the "
    "provider's name or roughly when the appointment was?"
)

INJECTION_REFUSAL_TEMPLATE = (
    "I can only help with scheduling appointments here. "
    "Is there an appointment I can help you with?"
)

PHI_REDIRECT_TEMPLATE = (
    "For your privacy, please don't share specific health "
    "details, account numbers, or other personal information "
    "in this chat. I just need your name, date of birth, "
    "and a confirmation factor to find your record."
)

BOOKING_CLAIM_FAILED_TEMPLATE = (
    "Something went wrong while I was trying to confirm "
    "that booking. Let me get you to our scheduling team "
    "at 555-0150 so we can get it sorted out."
)

# --- Prompt-Injection Patterns ---
INJECTION_PATTERNS = [
    r"ignore (all |any |the )?(previous|prior|above) (instructions|messages|prompts)",
    r"disregard (all |any |the )?(previous|prior|above) (instructions|messages|prompts)",
    r"you are now (a |an )?",
    r"forget (all |any |everything )?(you|your)",
    r"system (prompt|message|instruction)",
    r"reveal (your|the) (prompt|instructions|system)",
    r"act as (a |an )?",
    r"pretend (to be|you are)",
    r"new instructions:",
]

# --- PHI Detection Patterns ---
PHI_PATTERNS = {
    "ssn_like":     re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    "account_long": re.compile(r"\b\d{9,16}\b"),
    "mrn_prefix":   re.compile(r"\bMRN\s*[:#]?\s*\d{4,}\b",
                                re.IGNORECASE),
}

def _to_decimal(obj):
    """
    Recursively convert floats to Decimal for DynamoDB. DynamoDB
    rejects Python float values; every numeric field has to pass
    through Decimal on the way in. This is a recurring SDK gotcha.
    """
    if isinstance(obj, float):
        return Decimal(str(obj))
    if isinstance(obj, dict):
        return {k: _to_decimal(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_to_decimal(v) for v in obj]
    return obj

def _from_decimal(obj):
    """Inverse of `_to_decimal` for JSON serialization."""
    if isinstance(obj, Decimal):
        if obj % 1 == 0:
            return int(obj)
        return float(obj)
    if isinstance(obj, dict):
        return {k: _from_decimal(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_from_decimal(v) for v in obj]
    return obj

def _now_iso() -> str:
    """Current UTC time in ISO-8601 format."""
    return datetime.now(timezone.utc).isoformat()

def _parse_json_response(raw_text: str) -> dict:
    """
    Parse JSON from a model response, stripping common markdown
    wrappers. Models sometimes wrap JSON in fenced code blocks
    even when told not to; defensive parsing keeps the pipeline
    robust to that.
    """
    cleaned = raw_text.strip()
    if cleaned.startswith("```json"):
        cleaned = cleaned[7:]
    if cleaned.startswith("```"):
        cleaned = cleaned[3:]
    if cleaned.endswith("```"):
        cleaned = cleaned[:-3]
    try:
        return json.loads(cleaned.strip())
    except json.JSONDecodeError:
        logger.warning(
            "Failed to parse JSON from model; returning empty")
        return {}

def _redact_pii_for_logging(text: str) -> str:
    """
    Light redaction for log lines. The audit pipeline gets the
    original; only logs get the redaction.
    """
    redacted = text
    for pattern in PHI_PATTERNS.values():
        redacted = pattern.sub("[REDACTED]", redacted)
    return redacted

def _emit_event(detail_type: str, detail: dict) -> None:
    """
    Emit an EventBridge event. Wrapped in try/except so a transient
    EventBridge failure does not block the chat-handler response.
    """
    try:
        eventbridge_client.put_events(Entries=[{
            "Source":       "scheduling_bot",
            "DetailType":   detail_type,
            "Detail":       json.dumps(detail),
            "EventBusName": SCHEDULING_EVENT_BUS_NAME,
        }])
    except Exception as exc:
        logger.error(
            "EventBridge emit failed for %s: %s",
            detail_type, exc)

def _put_metric(metric_name: str, value: float,
                dimensions: dict) -> None:
    """Emit a CloudWatch metric. Best-effort; never blocks the response."""
    try:
        cloudwatch_client.put_metric_data(
            Namespace=CLOUDWATCH_NAMESPACE,
            MetricData=[{
                "MetricName": metric_name,
                "Value":      value,
                "Unit":       "Count",
                "Dimensions": [
                    {"Name": k, "Value": str(v)}
                    for k, v in dimensions.items()
                ],
            }])
    except Exception as exc:
        logger.error(
            "CloudWatch put_metric_data failed for %s: %s",
            metric_name, exc)

def _audit_tool_call(session_id: str,
                      tool: str,
                      arguments: dict,
                      result_summary: dict,
                      latency_ms: int,
                      outcome: str) -> None:
    """
    Persist a tool-call ledger entry. Every tool invocation gets a
    durable record: which tool, what arguments (redacted of
    sensitive fields), what outcome, what latency. The ledger is
    queryable for auditing; the per-conversation audit record at
    close time pulls all ledger entries for the session.
    """
    try:
        table = dynamodb.Table(TOOL_CALL_LEDGER_TABLE)
        table.put_item(Item=_to_decimal({
            "session_id":      session_id,
            "invoked_at":      _now_iso(),
            "tool":            tool,
            "arguments_summary": _redact_tool_args(arguments),
            "result_summary":  result_summary,
            "latency_ms":      latency_ms,
            "outcome":         outcome,
        }))
    except Exception as exc:
        logger.error(
            "Tool-call ledger write failed for %s/%s: %s",
            session_id, tool, exc)

def _redact_tool_args(arguments: dict) -> dict:
    """
    Strip sensitive fields from tool arguments before they go in
    the ledger. The ledger keeps enough to reconstruct what
    happened without storing raw identifiers.
    """
    redacted = dict(arguments)
    sensitive_keys = {
        "name", "date_of_birth", "confirmation_factor",
        "phone", "ssn", "mrn", "patient_id",
    }
    for key in list(redacted.keys()):
        if key in sensitive_keys:
            redacted[key] = "[REDACTED]"
    return redacted

def receive_message(channel: str,
                    channel_session_id: str,
                    user_message: str,
                    auth_context: Optional[dict] = None,
                    language: str = "en-US") -> dict:
    """
    Entry point for an inbound chat message.

    Args:
        channel:             Channel identifier (web_chat, in_app,
                             portal_embed, sms, voice).
        channel_session_id:  Stable identifier from the channel.
        user_message:        The patient's typed message.
        auth_context:        Optional dict with `authenticated`
                             (bool) and `patient_id` (str) when the
                             patient arrives through an authenticated
                             portal-embed channel.
        language:            Detected or declared language.

    Returns:
        A dict with the response to send to the user.
    """
    auth_context = auth_context or {"authenticated": False}

    # Step 1A: identify or create the session.
    session = _get_or_create_session(
        channel=channel,
        channel_session_id=channel_session_id,
        auth_context=auth_context,
        language=language)
    session_id = session["session_id"]

    # Step 1B: on the first message, send the greeting and
    # disclosure. The disclosure tells the patient this is a
    # chatbot, what scheduling actions it can do, and how to
    # reach a human.
    attach_greeting = (session["message_count"] == 0)
    if attach_greeting:
        _emit_event("conversation_started", {
            "session_id": session_id,
            "channel":    channel,
            "language":   language,
            "authenticated":
                bool(auth_context.get("authenticated")),
        })

    # Step 1C: persist the user's turn.
    _append_turn(
        session_id=session_id,
        turn={
            "speaker":   "user",
            "text":      user_message,
            "timestamp": _now_iso(),
            "language":  language,
        })

    # Step 1D: run the input-screening primitive (same as
    # recipe 11.1).
    screening_result = _screen_input(
        session_id=session_id,
        user_message=user_message,
        language=language)

    if screening_result["action"] != "proceed":
        return _handle_screening_action(
            session_id=session_id,
            channel=channel,
            screening_result=screening_result,
            attach_greeting=attach_greeting,
            language=language)

    # Step 1E: continue to intent classification.
    return _handle_in_scope_message(
        session_id=session_id,
        channel=channel,
        user_message=user_message,
        auth_context=auth_context,
        attach_greeting=attach_greeting,
        language=language)

def _get_or_create_session(channel: str,
                            channel_session_id: str,
                            auth_context: dict,
                            language: str) -> dict:
    """Look up the active conversation or create a new one."""
    session_key = f"{channel}#{channel_session_id}"
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)

    try:
        response = table.get_item(
            Key={"session_key": session_key})
    except Exception as exc:
        logger.warning(
            "Conversation state lookup failed: %s", exc)
        response = {}

    item = response.get("Item")
    if item:
        item["message_count"] = (
            int(item.get("message_count", 0)) + 1)
        table.put_item(Item=_to_decimal(item))
        return _from_decimal(item)

    # Brand-new session.
    new_session = {
        "session_key":         session_key,
        "session_id":          str(uuid.uuid4()),
        "channel":              channel,
        "channel_session_id":  channel_session_id,
        "language":            language,
        "started_at":          _now_iso(),
        "message_count":       0,
        "auth_context":        auth_context,
        "verified_patient_id": (
            auth_context.get("patient_id")
            if auth_context.get("authenticated") else None),
        "assurance_level":  (
            "authenticated"
            if auth_context.get("authenticated") else None),
        # Versions stamped per session for audit reproducibility.
        "prompt_version":          PROMPT_VERSION,
        "agent_version":           AGENT_VERSION,
        "kb_id":                   KNOWLEDGE_BASE_ID,
        "guardrail_id":            GUARDRAIL_ID,
        "guardrail_version":       GUARDRAIL_VERSION,
        "model_id":                ORCHESTRATION_MODEL_ID,
        "visit_type_catalog_version":
            VISIT_TYPE_CATALOG_VERSION,
        "scheduling_policy_version":
            SCHEDULING_POLICY_VERSION,
        # Conversation-level state filled in as we go.
        "intent":               None,
        "search_parameters":    {},
        "last_candidates":      [],
        "held_slot":            None,
        "hold_id":              None,
        "crisis_detected":      False,
        "crisis_severity":      None,
        "scope_violation_count": 0,
        "bookings_completed":   0,
        "reschedules_completed": 0,
        "cancellations_completed": 0,
        "handoffs_offered":     0,
        "handoffs_accepted":    0,
        "feedback_history":     [],
    }
    table.put_item(Item=_to_decimal(new_session))
    return new_session

def _append_turn(session_id: str, turn: dict) -> None:
    """Append a turn record to the conversation-metadata table."""
    table = dynamodb.Table(CONVERSATION_METADATA_TABLE)
    item = {
        "session_id": session_id,
        "timestamp":  turn["timestamp"],
        **turn,
    }
    try:
        table.put_item(Item=_to_decimal(item))
    except Exception as exc:
        logger.error(
            "Failed to append turn for %s: %s", session_id, exc)

def _recent_turns(session_id: str, k: int = 4) -> list:
    """Return the most recent k turns for context."""
    table = dynamodb.Table(CONVERSATION_METADATA_TABLE)
    try:
        response = table.query(
            KeyConditionExpression=
                boto3.dynamodb.conditions.Key("session_id")
                    .eq(session_id),
            ScanIndexForward=False,
            Limit=k)
    except Exception as exc:
        logger.warning(
            "Recent-turns query failed for %s: %s",
            session_id, exc)
        return []
    items = [_from_decimal(i) for i in response.get("Items", [])]
    return list(reversed(items))

def _screen_input(session_id: str,
                   user_message: str,
                   language: str) -> dict:
    """
    Run the input-screening pass: crisis detection, prompt
    injection detection, and PHI minimization.

    Crisis detection is highest priority and preempts everything.
    A patient describing chest pain while asking about a follow-up
    must not have that signal lost in the booking flow.
    """
    # Crisis detection. The scheduling bot does not attempt to
    # handle the crisis itself; it routes to triage.
    lowered = user_message.lower()
    for cue in HIGH_ACUITY_CUES:
        if cue in lowered:
            return {
                "action":   "crisis_response",
                "matched_cue": cue,
            }

    # Prompt-injection detection.
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, lowered):
            return {
                "action":  "injection_refusal",
                "pattern": pattern,
            }

    # PHI minimization. Before identity verification, the bot does
    # not need PHI; if the user volunteers something sensitive,
    # flag for redaction and gently redirect.
    matched = []
    for category, pattern in PHI_PATTERNS.items():
        if pattern.search(user_message):
            matched.append(category)
    if matched:
        return {
            "action":         "phi_redirect",
            "phi_categories": matched,
        }

    return {"action": "proceed"}

def _handle_screening_action(session_id: str,
                              channel: str,
                              screening_result: dict,
                              attach_greeting: bool,
                              language: str) -> dict:
    """Build response for a screening action that did not pass."""
    action = screening_result["action"]

    if action == "crisis_response":
        response_text = CRISIS_ROUTE_TEMPLATE
        _update_session_flag(
            session_id, "crisis_detected", True)
        _emit_event("crisis_detected", {
            "session_id":  session_id,
            "matched_cue": screening_result["matched_cue"],
        })
        _put_metric("CrisisDetected", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      response_text,
            "timestamp": _now_iso(),
            "language":  language,
            "screening_action": "crisis_response",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=response_text,
            attach_greeting=attach_greeting,
            disposition="crisis_routed")

    if action == "injection_refusal":
        _put_metric("InjectionAttemptDetected", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      INJECTION_REFUSAL_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "screening_action": "injection_refusal",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=INJECTION_REFUSAL_TEMPLATE,
            attach_greeting=attach_greeting,
            disposition="continued")

    if action == "phi_redirect":
        _put_metric("PHIVolunteeredByUser", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      PHI_REDIRECT_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "screening_action": "phi_redirect",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=PHI_REDIRECT_TEMPLATE,
            attach_greeting=attach_greeting,
            disposition="continued")

    raise ValueError(f"Unknown screening action: {action}")

def _update_session_flag(session_id: str,
                          flag_name: str,
                          value) -> None:
    """Update a single field on the active session row."""
    # In production, resolve the partition key (session_key) from
    # session_id via a GSI on session_id rather than constructing
    # a synthetic key. The demo's MockTable is permissive about this.
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    try:
        # Look up by session_id index in production. The demo helper
        # below scans the in-memory table; production uses Query on
        # a GSI keyed by session_id.
        session_key = _resolve_session_key(session_id)
        if session_key is None:
            logger.warning(
                "No session_key found for session_id=%s", session_id)
            return
        table.update_item(
            Key={"session_key": session_key},
            UpdateExpression="SET #f = :v",
            ExpressionAttributeNames={"#f": flag_name},
            ExpressionAttributeValues={":v": _to_decimal(value)})
    except Exception as exc:
        logger.warning(
            "Failed to set %s on session %s: %s",
            flag_name, session_id, exc)

def _resolve_session_key(session_id: str) -> Optional[str]:
    """
    Resolve a session_id to the session_key partition key. In
    production this is a Query on a GSI; the demo's MockTable
    exposes a helper to scan the in-memory items.
    """
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    # If the underlying table is the demo mock, it has a helper.
    if hasattr(table, "_find_session_key_by_session_id"):
        return table._find_session_key_by_session_id(session_id)
    # Production path: query the GSI on session_id.
    try:
        response = table.query(
            IndexName="session_id-index",
            KeyConditionExpression=
                boto3.dynamodb.conditions.Key("session_id")
                    .eq(session_id),
            Limit=1)
        items = response.get("Items", [])
        if items:
            return items[0].get("session_key")
    except Exception as exc:
        logger.warning(
            "session_key resolution failed for %s: %s",
            session_id, exc)
    return None

def _build_chat_reply(session_id: str,
                       response_text: str,
                       attach_greeting: bool,
                       disposition: str) -> dict:
    """Build the user-facing chat reply payload."""
    full_text = response_text
    if attach_greeting:
        full_text = (
            f"{GREETING_AND_DISCLOSURE_TEMPLATE}\n\n{full_text}")
    return {
        "session_id":  session_id,
        "response":    full_text,
        "disposition": disposition,
    }

def _handle_in_scope_message(session_id: str,
                              channel: str,
                              user_message: str,
                              auth_context: dict,
                              attach_greeting: bool,
                              language: str) -> dict:
    """
    Classify the user's intent and route accordingly: a clarifying
    question if confidence is low, an out-of-scope handoff, or
    continuation into the scheduling flow.
    """
    classification = _classify_scheduling_intent(
        session_id=session_id,
        user_message=user_message,
        language=language)

    intent = classification.get("intent")
    confidence = classification.get("confidence",
                                     Decimal("0.0"))

    # Low confidence: clarify.
    if confidence < INTENT_CONFIDENCE_THRESHOLD:
        clarification = (
            "I want to make sure I help with the right thing. "
            "Are you looking to book a new appointment, "
            "reschedule, cancel, or check an existing one?"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      clarification,
            "timestamp": _now_iso(),
            "language":  language,
            "intent_action": "clarify",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=clarification,
            attach_greeting=attach_greeting,
            disposition="clarification_requested")

    # Out-of-scope handoffs.
    if intent in OUT_OF_SCOPE_INTENTS:
        target = OUT_OF_SCOPE_INTENTS[intent]
        response_text = OUT_OF_SCOPE_HANDOFFS.get(
            intent, OUT_OF_SCOPE_HANDOFFS["out_of_scope"])
        _put_metric("HandoffOffered", 1, {
            "channel":  channel,
            "intent":   intent,
            "target":   target,
            "language": language,
        })
        _emit_event("handoff_offered", {
            "session_id": session_id,
            "intent":     intent,
            "target":     target,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      response_text,
            "timestamp": _now_iso(),
            "language":  language,
            "intent_action": "handoff",
            "intent":        intent,
            "target":        target,
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=response_text,
            attach_greeting=attach_greeting,
            disposition="handoff_offered")

    # In-scope: stamp intent on session and route to the
    # scheduling flow (Step 3 onwards).
    if intent in SCHEDULING_INTENTS:
        _update_session_flag(session_id, "intent", intent)
        _update_session_flag(
            session_id,
            "search_parameters",
            classification.get("extracted_parameters", {}))
        return _route_to_scheduling_flow(
            session_id=session_id,
            channel=channel,
            user_message=user_message,
            intent=intent,
            extracted_parameters=
                classification.get("extracted_parameters", {}),
            auth_context=auth_context,
            attach_greeting=attach_greeting,
            language=language)

    # Unknown intent label from the classifier.
    return _build_chat_reply(
        session_id=session_id,
        response_text=(
            "I'm not sure I caught that. Are you looking to "
            "book, reschedule, cancel, or check an "
            "appointment?"),
        attach_greeting=attach_greeting,
        disposition="clarification_requested")

def _classify_scheduling_intent(session_id: str,
                                 user_message: str,
                                 language: str) -> dict:
    """
    Classify the user's message into a scheduling intent and
    extract any preliminary parameters (provider, time-window,
    insurance, reason for visit).
    """
    in_scope_csv = ", ".join(SCHEDULING_INTENTS)
    out_of_scope_csv = ", ".join(OUT_OF_SCOPE_INTENTS.keys())
    recent = _recent_turns(session_id, k=4)
    history_text = "\n".join(
        f"{t['speaker']}: {t['text']}" for t in recent
    )

    system_prompt = (
        "You classify patient messages for a healthcare "
        "appointment-scheduling bot.\n\n"
        "Return ONLY valid JSON in this exact shape:\n"
        "{\n"
        '  "intent": "<one of the categories below>",\n'
        '  "confidence": <number between 0 and 1>,\n'
        '  "extracted_parameters": {\n'
        '    "provider_hint": "<provider name or null>",\n'
        '    "specialty": "<specialty name or null>",\n'
        '    "reason_for_visit": "<short text or null>",\n'
        '    "time_window": "<short text or null>",\n'
        '    "insurance_plan": "<plan name or null>",\n'
        '    "appointment_descriptor": "<for reschedule or '
        'cancel: text describing which existing appointment, '
        'or null>"\n'
        '  },\n'
        '  "reasoning": "<one short sentence>"\n'
        "}\n\n"
        f"IN-SCOPE INTENTS: {in_scope_csv}\n"
        f"OUT-OF-SCOPE INTENTS: {out_of_scope_csv}\n\n"
        "RULES:\n"
        "- A clinical question (symptoms, conditions, dosing, "
        "should-I-come-in) is ALWAYS clinical_question.\n"
        "- A request to refill a prescription is "
        "refill_request.\n"
        "- A plan-specific coverage question is "
        "benefits_eligibility.\n"
        "- General clinic questions (parking, hours) are "
        "general_question.\n"
        "- Any scheduling-related action goes to the "
        "appropriate scheduling intent above."
    )
    user_prompt = (
        f"RECENT CONVERSATION:\n{history_text}\n\n"
        f"NEW USER MESSAGE: {user_message}"
    )

    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens":  400,
        "temperature": 0.0,
        "system":      system_prompt,
        "messages": [{
            "role":    "user",
            "content": user_prompt,
        }],
    })

    try:
        response = bedrock_runtime.invoke_model(
            modelId=INTENT_CLASSIFIER_MODEL_ID,
            contentType="application/json",
            accept="application/json",
            body=body)
        payload = json.loads(response["body"].read())
        parsed = _parse_json_response(
            payload["content"][0]["text"])
    except Exception as exc:
        logger.warning(
            "Intent classifier failed: %s", exc)
        return {
            "intent":     "out_of_scope",
            "confidence": Decimal("0.0"),
            "extracted_parameters": {},
        }

    return {
        "intent":     parsed.get("intent", "out_of_scope"),
        "confidence":
            Decimal(str(parsed.get("confidence", 0))),
        "extracted_parameters":
            parsed.get("extracted_parameters", {}),
    }

def _route_to_scheduling_flow(session_id: str,
                               channel: str,
                               user_message: str,
                               intent: str,
                               extracted_parameters: dict,
                               auth_context: dict,
                               attach_greeting: bool,
                               language: str) -> dict:
    """
    The scheduling-specific flow needs identity verification before
    any action that touches the patient's record.
    """
    # Determine the required assurance level for this (intent,
    # auth state, optionally hours-to-existing-appointment).
    policy_ctx = {
        "authenticated": auth_context.get("authenticated", False),
        "hours_to_appointment": 999,  # default; refined later
    }
    required_assurance = _required_assurance_for(intent, policy_ctx)

    # Authenticated short-circuit.
    if (auth_context.get("authenticated")
            and required_assurance == "authenticated"):
        _update_session_flag(
            session_id, "verified_patient_id",
            auth_context.get("patient_id"))
        _update_session_flag(
            session_id, "assurance_level", "authenticated")
        return _continue_after_identity(
            session_id=session_id,
            channel=channel,
            intent=intent,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language)

    # Otherwise, gather identifiers conversationally. The demo
    # collects them all in one ask; production uses a multi-turn
    # collection flow with backtracking.
    identifiers = _collect_identifiers_from_message(
        user_message=user_message,
        recent_turns=_recent_turns(session_id, k=4))

    if not identifiers["complete"]:
        ask_text = (
            "I just need to verify it's you first. Can I get "
            "your name, date of birth, and the last four "
            "digits of the phone number we have on file?"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      ask_text,
            "timestamp": _now_iso(),
            "language":  language,
            "identity_action": "ask_for_identifiers",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=ask_text,
            attach_greeting=attach_greeting,
            disposition="awaiting_identity")

    # Invoke the patient_lookup tool.
    lookup_start = datetime.now(timezone.utc)
    lookup_result = patient_lookup_tool(
        name=identifiers["name"],
        date_of_birth=identifiers["date_of_birth"],
        confirmation_factor=identifiers["confirmation_factor"])
    lookup_latency_ms = int(
        (datetime.now(timezone.utc) - lookup_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="patient_lookup",
        arguments=identifiers,
        result_summary={
            "match_count": lookup_result.get("match_count", 0),
            "confidence":
                str(lookup_result.get("confidence", 0)),
        },
        latency_ms=lookup_latency_ms,
        outcome=("verified"
                 if lookup_result.get("match_count") == 1
                 else "ambiguous_or_no_match"))

    match_count = lookup_result.get("match_count", 0)
    confidence = Decimal(str(lookup_result.get("confidence", 0)))

    if match_count == 0:
        no_match_text = (
            "I'm not finding a record matching that "
            "information. Are you a new patient, or could the "
            "name or date of birth be different from what we "
            "have on file? I can also connect you with our "
            "front desk at 555-0100."
        )
        _put_metric("IdentityVerificationFailed", 1, {
            "channel":  channel,
            "reason":   "no_match",
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      no_match_text,
            "timestamp": _now_iso(),
            "language":  language,
            "identity_action": "no_match",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=no_match_text,
            attach_greeting=attach_greeting,
            disposition="identity_no_match")

    if match_count > 1:
        ambiguous_text = (
            "I see more than one record that could match. "
            "Could you also share the ZIP code we have on "
            "file?"
        )
        _put_metric("IdentityVerificationAmbiguous", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      ambiguous_text,
            "timestamp": _now_iso(),
            "language":  language,
            "identity_action": "ambiguous_match",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=ambiguous_text,
            attach_greeting=attach_greeting,
            disposition="identity_ambiguous")

    # Single match: check confidence against assurance threshold.
    threshold = ASSURANCE_MATCH_THRESHOLDS[required_assurance]
    if confidence < threshold:
        step_up_text = (
            "I want to make sure it's really you. I'll send a "
            "one-time code to the phone number on file. Could "
            "you read it back to me when you get it?"
        )
        _put_metric("IdentityStepUpRequired", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      step_up_text,
            "timestamp": _now_iso(),
            "language":  language,
            "identity_action": "step_up_requested",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=step_up_text,
            attach_greeting=attach_greeting,
            disposition="identity_step_up")

    # Verified. Stamp on session and continue.
    _update_session_flag(
        session_id, "verified_patient_id",
        lookup_result["patient_id"])
    _update_session_flag(
        session_id, "assurance_level",
        required_assurance)
    _put_metric("IdentityVerified", 1, {
        "channel":   channel,
        "assurance": required_assurance,
        "language":  language,
    })

    return _continue_after_identity(
        session_id=session_id,
        channel=channel,
        intent=intent,
        extracted_parameters=extracted_parameters,
        attach_greeting=attach_greeting,
        language=language,
        verified_just_now=True,
        patient_first_name=lookup_result.get("first_name"))

def _required_assurance_for(intent: str, ctx: dict) -> str:
    """
    Look up the required assurance level for (intent, ctx) using
    the policy table. Falls back to "basic".
    """
    rules = IDENTITY_POLICY.get(intent, [(lambda c: True, "basic")])
    for predicate, level in rules:
        if predicate(ctx):
            return level
    return "basic"

def _collect_identifiers_from_message(user_message: str,
                                        recent_turns: list) -> dict:
    """
    Pull name, DOB, and a confirmation factor out of the user's
    recent messages. The demo uses simple regex; production uses
    the LLM with the recent conversation as context to extract
    identifiers conversationally.
    """
    combined = user_message + " " + " ".join(
        t.get("text", "")
        for t in recent_turns
        if t.get("speaker") == "user")

    name_match = re.search(
        r"(?:my name is|i'?m|this is)\s+"
        r"([A-Z][a-z]+(?:\s+[A-Z][a-z]+)+)",
        combined)
    name = name_match.group(1) if name_match else None
    # Allow a bare "First Last" if name pattern not found.
    if not name:
        bare_name = re.search(
            r"\b([A-Z][a-z]+\s+[A-Z][a-z]+)\b", combined)
        name = bare_name.group(1) if bare_name else None

    dob_match = re.search(
        r"\b(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{2,4})\b",
        combined)
    dob = dob_match.group(1) if dob_match else None

    # Confirmation factor: a 4-digit number not part of a longer
    # number sequence (heuristic for last-four-of-phone).
    conf_match = re.search(
        r"(?<!\d)(\d{4})(?!\d)", combined)
    confirmation = conf_match.group(1) if conf_match else None

    complete = bool(name and dob and confirmation)
    return {
        "name":                 name,
        "date_of_birth":        dob,
        "confirmation_factor": confirmation,
        "complete":             complete,
    }

def _continue_after_identity(session_id: str,
                              channel: str,
                              intent: str,
                              extracted_parameters: dict,
                              attach_greeting: bool,
                              language: str,
                              verified_just_now: bool = False,
                              patient_first_name: Optional[str] = None
                              ) -> dict:
    """
    Branch on intent now that identity is verified.

    new_appointment goes to slot search; reschedule_appointment and
    cancel_appointment go to the existing-appointment lookup;
    check_appointment goes to a read-only lookup; update_preferences
    is out of scope for this demo and falls through to a handoff.
    """
    if intent == "new_appointment":
        return _search_for_slots(
            session_id=session_id,
            channel=channel,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language,
            verified_just_now=verified_just_now,
            patient_first_name=patient_first_name)

    if intent in ("reschedule_appointment",
                  "cancel_appointment"):
        return _handle_reschedule_or_cancel(
            session_id=session_id,
            channel=channel,
            intent=intent,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language)

    if intent == "check_appointment":
        return _check_appointment(
            session_id=session_id,
            channel=channel,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language)

    # update_preferences and anything else falls back to a handoff.
    return _build_chat_reply(
        session_id=session_id,
        response_text=OUT_OF_SCOPE_HANDOFFS["out_of_scope"],
        attach_greeting=attach_greeting,
        disposition="handoff_offered")

def _search_for_slots(session_id: str,
                       channel: str,
                       extracted_parameters: dict,
                       attach_greeting: bool,
                       language: str,
                       verified_just_now: bool,
                       patient_first_name: Optional[str]) -> dict:
    """Map reason-for-visit to a visit type, then call slot_search."""
    parameters = dict(extracted_parameters)

    # Step 4A: optional clinical-content detection on reason-for-visit.
    # If the patient describes high-acuity symptoms in the reason
    # field, route to triage rather than self-scheduling. (The
    # input-screening pass in Step 1 already catches the most
    # obvious cases; this is the secondary check on the structured
    # reason field specifically.)
    reason = (parameters.get("reason_for_visit") or "").lower()
    for cue in HIGH_ACUITY_CUES:
        if cue in reason:
            triage_text = (
                "Based on what you described, I'd rather have "
                "a nurse decide the right kind of visit for "
                "you. Let me connect you with our nurse "
                "advice line at 555-0100."
            )
            _emit_event("high_acuity_routed", {
                "session_id": session_id,
                "matched_cue": cue,
            })
            _append_turn(session_id, {
                "speaker":   "assistant",
                "text":      triage_text,
                "timestamp": _now_iso(),
                "language":  language,
                "scheduling_action": "triage_routed",
            })
            return _build_chat_reply(
                session_id=session_id,
                response_text=triage_text,
                attach_greeting=attach_greeting,
                disposition="triage_routed")

    # Step 4B: visit-type mapping. The mapping uses the LLM with
    # the institutional visit-type catalog as context. The demo
    # uses a simpler keyword-overlap heuristic against the catalog.
    if not parameters.get("visit_type"):
        mapping = _map_reason_to_visit_type(
            reason=parameters.get("reason_for_visit", ""),
            specialty_hint=parameters.get("specialty"))
        if mapping["confidence"] < VISIT_TYPE_CONFIDENCE_THRESHOLD:
            # Ask the patient to clarify.
            top = mapping.get("top_candidates", [])
            top_descriptions = "\n".join(
                f"- {VISIT_TYPE_CATALOG[v]['display_name']}"
                for v in top
                if v in VISIT_TYPE_CATALOG)
            clarify_text = (
                "A couple of visit types could fit what you "
                "described. Which sounds right?\n"
                f"{top_descriptions}"
            )
            _append_turn(session_id, {
                "speaker":   "assistant",
                "text":      clarify_text,
                "timestamp": _now_iso(),
                "language":  language,
                "scheduling_action": "clarify_visit_type",
            })
            return _build_chat_reply(
                session_id=session_id,
                response_text=clarify_text,
                attach_greeting=attach_greeting,
                disposition="clarify_visit_type")

        parameters["visit_type"] = mapping["visit_type"]

    # Step 4C: visit-type policy check. Some visit types are not
    # self-schedulable (e.g., new patient cardiology requires
    # triage first).
    vt_config = VISIT_TYPE_CATALOG.get(parameters["visit_type"], {})
    if not vt_config.get("self_schedulable", True):
        not_self_text = (
            f"For a "
            f"{vt_config.get('display_name', 'visit')}, our "
            "scheduling team needs to do a quick intake call "
            "first so we can match you with the right "
            "provider. They can be reached at 555-0150, or I "
            "can have them call you. Which would you prefer?"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      not_self_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "not_self_schedulable",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=not_self_text,
            attach_greeting=attach_greeting,
            disposition="not_self_schedulable")

    # Step 4D: invoke slot_search.
    search_start = datetime.now(timezone.utc)
    search_result = slot_search_tool(
        patient_id=_session_patient_id(session_id),
        provider_hint=parameters.get("provider_hint"),
        visit_type=parameters["visit_type"],
        time_window=parameters.get("time_window"),
        insurance_plan=parameters.get("insurance_plan"))
    search_latency_ms = int(
        (datetime.now(timezone.utc) - search_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="slot_search",
        arguments=parameters,
        result_summary={
            "slot_count": len(search_result.get("slots", [])),
            "top_slot_provider": (
                search_result.get("slots", [{}])[0].get("provider")
                if search_result.get("slots") else None),
        },
        latency_ms=search_latency_ms,
        outcome=("results" if search_result.get("slots")
                 else "no_results"))

    slots = search_result.get("slots", [])

    # Step 4E: no-results handling. Be honest, offer alternatives.
    if not slots:
        _put_metric("SlotSearchNoResults", 1, {
            "channel":    channel,
            "visit_type": parameters["visit_type"],
            "language":   language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      NO_SLOTS_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "no_slots_available",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=NO_SLOTS_TEMPLATE,
            attach_greeting=attach_greeting,
            disposition="no_slots_available")

    # Step 4F: present top 2-3 candidates. Stash them on the
    # session so the next user turn can resolve a refinement
    # ("the Tuesday one" or "anything earlier?").
    top_candidates = slots[:3]
    _update_session_flag(
        session_id, "last_candidates", top_candidates)
    _update_session_flag(
        session_id, "search_parameters", parameters)

    response_text = _render_candidates(
        candidates=top_candidates,
        verified_just_now=verified_just_now,
        patient_first_name=patient_first_name)

    _append_turn(session_id, {
        "speaker":   "assistant",
        "text":      response_text,
        "timestamp": _now_iso(),
        "language":  language,
        "scheduling_action": "offer_slots",
        "candidate_slot_ids": [
            s["slot_id"] for s in top_candidates],
    })
    return _build_chat_reply(
        session_id=session_id,
        response_text=response_text,
        attach_greeting=attach_greeting,
        disposition="slots_offered")

def _map_reason_to_visit_type(reason: str,
                                specialty_hint: Optional[str]
                                ) -> dict:
    """
    Map a natural-language reason for visit to an institutional
    visit type. The demo uses keyword overlap against the catalog's
    natural_language_cues list. Production uses the LLM with the
    catalog descriptions as context, plus the patient's
    care-team and prior-visit context.
    """
    reason_lower = reason.lower()
    scores = []
    for vt_id, vt in VISIT_TYPE_CATALOG.items():
        if (specialty_hint
                and vt.get("specialty") != specialty_hint.lower()):
            continue
        score = sum(
            1
            for cue in vt["natural_language_cues"]
            if cue in reason_lower)
        if score > 0:
            scores.append((vt_id, score))

    if not scores:
        # Fallback: default to a primary care follow-up if nothing
        # else matches. Production should not have a silent
        # default; the demo keeps it simple.
        return {
            "visit_type": "primary_care_followup_20min",
            "confidence": Decimal("0.3"),
            "top_candidates": list(VISIT_TYPE_CATALOG.keys())[:3],
        }

    scores.sort(key=lambda x: x[1], reverse=True)
    top_vt, top_score = scores[0]
    # Confidence proxy: ratio of top score to total.
    total = sum(s for _, s in scores)
    confidence = Decimal(str(top_score / max(total, 1)))
    return {
        "visit_type":      top_vt,
        "confidence":      confidence,
        "top_candidates":  [vt for vt, _ in scores[:3]],
    }

def _session_patient_id(session_id: str) -> Optional[str]:
    """Read verified_patient_id from the session state row."""
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    session_key = _resolve_session_key(session_id)
    if not session_key:
        return None
    response = table.get_item(Key={"session_key": session_key})
    item = response.get("Item")
    if not item:
        return None
    return item.get("verified_patient_id")

def _render_candidates(candidates: list,
                        verified_just_now: bool,
                        patient_first_name: Optional[str]) -> str:
    """Render 2-3 candidate slots in plain English."""
    lines = []
    if verified_just_now and patient_first_name:
        lines.append(f"Thanks {patient_first_name}, you're verified.")
    elif verified_just_now:
        lines.append("Perfect, you're verified.")

    if len(candidates) == 1:
        c = candidates[0]
        lines.append(
            f"I see one opening: {_format_slot_text(c)}. "
            f"Want to grab it?")
    else:
        lines.append(
            "I see a few openings. "
            "Which works best?")
        for c in candidates:
            lines.append(f"- {_format_slot_text(c)}")
    return "\n".join(lines)

def _format_slot_text(slot: dict) -> str:
    """Format a slot for plain-English presentation."""
    return (
        f"{slot.get('display_time')} with "
        f"Dr. {slot.get('provider_last_name', slot.get('provider'))} "
        f"at {slot.get('location_name', 'the clinic')}")

def handle_slot_response(session_id: str,
                          channel: str,
                          user_message: str,
                          language: str = "en-US") -> dict:
    """
    Entry point for follow-up turns once slots are on the table.
    The chat handler routes here when the session's last
    disposition was "slots_offered".

    Classifies the user's response as refine / select / cancel
    and acts accordingly.
    """
    session = _session_state(session_id)
    candidates = session.get("last_candidates") or []
    parameters = session.get("search_parameters") or {}

    classification = _classify_slot_response(
        user_message=user_message,
        candidates=candidates)

    if classification["action"] == "refine":
        # Apply the refinement and re-search.
        refined_params = _apply_refinement(
            current=parameters,
            refinement=classification["refinement"])
        return _search_for_slots(
            session_id=session_id,
            channel=channel,
            extracted_parameters=refined_params,
            attach_greeting=False,
            language=language,
            verified_just_now=False,
            patient_first_name=None)

    if classification["action"] == "select":
        chosen = candidates[classification["choice_index"]]
        return _place_hold_and_confirm(
            session_id=session_id,
            channel=channel,
            chosen_slot=chosen,
            language=language)

    if classification["action"] == "cancel":
        bye_text = (
            "No problem. Let me know if you'd like to try "
            "again. Take care!"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      bye_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "abandon_search",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=bye_text,
            attach_greeting=False,
            disposition="abandoned")

    # Fallback: re-render candidates.
    return _build_chat_reply(
        session_id=session_id,
        response_text=_render_candidates(
            candidates=candidates,
            verified_just_now=False,
            patient_first_name=None),
        attach_greeting=False,
        disposition="slots_offered")

def _place_hold_and_confirm(session_id: str,
                              channel: str,
                              chosen_slot: dict,
                              language: str) -> dict:
    """
    Place a short-term hold on the chosen slot and ask the patient
    to confirm before booking. The hold is the safety net that
    prevents another channel from grabbing the slot during the
    confirmation step.
    """
    hold_start = datetime.now(timezone.utc)
    hold_result = slot_hold_tool(
        slot_id=chosen_slot["slot_id"],
        patient_id=_session_patient_id(session_id),
        ttl_seconds=HOLD_TTL_SECONDS)
    hold_latency_ms = int(
        (datetime.now(timezone.utc) - hold_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="slot_hold",
        arguments={"slot_id": chosen_slot["slot_id"]},
        result_summary={
            "hold_id":    hold_result.get("hold_id"),
            "expires_at": hold_result.get("expires_at"),
            "outcome":    hold_result.get("outcome"),
        },
        latency_ms=hold_latency_ms,
        outcome=hold_result.get("outcome", "unknown"))

    if hold_result.get("outcome") == "no_longer_available":
        # Slot was taken between search and hold attempt.
        _put_metric("SlotHoldRaceLost", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      SLOT_RACE_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "slot_no_longer_available",
        })
        # Re-search transparently after the apology.
        session = _session_state(session_id)
        return _search_for_slots(
            session_id=session_id,
            channel=channel,
            extracted_parameters=session.get(
                "search_parameters", {}),
            attach_greeting=False,
            language=language,
            verified_just_now=False,
            patient_first_name=None)

    if hold_result.get("outcome") != "held":
        _put_metric("SlotHoldFailed", 1, {
            "channel":  channel,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      HOLD_FAILED_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "hold_failed",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=HOLD_FAILED_TEMPLATE,
            attach_greeting=False,
            disposition="hold_failed")

    # Hold succeeded. Stash on session and ask for confirmation.
    _update_session_flag(session_id, "held_slot", chosen_slot)
    _update_session_flag(
        session_id, "hold_id", hold_result["hold_id"])

    confirmation_prompt = _build_confirmation_prompt(chosen_slot)
    _append_turn(session_id, {
        "speaker":   "assistant",
        "text":      confirmation_prompt,
        "timestamp": _now_iso(),
        "language":  language,
        "scheduling_action": "ask_for_confirmation",
        "held_slot_id": chosen_slot["slot_id"],
    })
    return _build_chat_reply(
        session_id=session_id,
        response_text=confirmation_prompt,
        attach_greeting=False,
        disposition="awaiting_confirmation")

def _classify_slot_response(user_message: str,
                              candidates: list) -> dict:
    """
    Decide whether the user's message refines, selects, or cancels.
    Production uses the LLM with the offered candidates and
    recent conversation as context. The demo uses keyword
    heuristics.
    """
    lowered = user_message.lower().strip()

    cancel_cues = [
        "cancel", "never mind", "nevermind", "forget it",
        "i'll call", "i'll just call",
    ]
    for cue in cancel_cues:
        if cue in lowered:
            return {"action": "cancel"}

    refine_cues = {
        "earlier":         {"time_window_modifier": "earlier"},
        "later":           {"time_window_modifier": "later"},
        "different day":   {"day_modifier": "any_other"},
        "different provider": {"provider_modifier": "any_other"},
        "telehealth":      {"modality": "telehealth"},
        "in person":       {"modality": "in_person"},
    }
    for cue, refinement in refine_cues.items():
        if cue in lowered:
            return {
                "action":    "refine",
                "refinement": refinement,
            }

    # Selection cues: "the tuesday one", "tuesday", first/second
    # /third, "1"/"2"/"3", "yes" with a single candidate.
    for idx, c in enumerate(candidates):
        cue = (c.get("display_time") or "").lower()
        if cue and cue.split()[0] in lowered:
            return {"action": "select", "choice_index": idx}

    if "first" in lowered or lowered.startswith("1"):
        return {"action": "select", "choice_index": 0}
    if "second" in lowered or lowered.startswith("2"):
        if len(candidates) >= 2:
            return {"action": "select", "choice_index": 1}
    if "third" in lowered or lowered.startswith("3"):
        if len(candidates) >= 3:
            return {"action": "select", "choice_index": 2}

    if (lowered in ("yes", "yes please", "sure", "ok", "okay")
            and len(candidates) == 1):
        return {"action": "select", "choice_index": 0}

    return {"action": "unknown"}

def _apply_refinement(current: dict, refinement: dict) -> dict:
    """Apply a refinement to the current search parameters."""
    refined = dict(current)
    if "time_window_modifier" in refinement:
        existing = refined.get("time_window") or ""
        refined["time_window"] = (
            f"{existing} {refinement['time_window_modifier']}"
        ).strip()
    if "day_modifier" in refinement:
        # In production we'd track which day was just rejected.
        refined["exclude_days"] = (
            refined.get("exclude_days", []) + ["last_offered_day"])
    if "provider_modifier" in refinement:
        refined["provider_hint"] = None
    if "modality" in refinement:
        refined["modality"] = refinement["modality"]
    return refined

def _build_confirmation_prompt(slot: dict) -> str:
    """Restate the proposed appointment and ask for confirmation."""
    visit_type = slot.get("visit_type")
    vt_config = VISIT_TYPE_CATALOG.get(visit_type, {})
    duration = vt_config.get("duration_minutes", 30)
    return (
        f"Holding {_format_slot_text(slot)}. Just to "
        f"confirm: this is a {duration}-minute "
        f"{vt_config.get('display_name', 'visit')}, and "
        f"you'll want to arrive about 10 minutes early "
        f"for check-in. Want me to book it?"
    )

def _session_state(session_id: str) -> dict:
    """Read the session-state row keyed by session_id."""
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    session_key = _resolve_session_key(session_id)
    if not session_key:
        return {}
    response = table.get_item(Key={"session_key": session_key})
    return _from_decimal(response.get("Item", {}))

def handle_confirmation_response(session_id: str,
                                   channel: str,
                                   user_message: str,
                                   language: str = "en-US"
                                   ) -> dict:
    """
    Entry point for the user's confirmation turn. The chat handler
    routes here when the session's last disposition was
    "awaiting_confirmation".
    """
    session = _session_state(session_id)
    held_slot = session.get("held_slot")
    hold_id = session.get("hold_id")

    confirmation = _classify_confirmation_response(
        user_message=user_message)

    if confirmation == "decline":
        # Release the hold and ask what to do next.
        slot_hold_tool(
            slot_id=held_slot["slot_id"]
            if held_slot else None,
            patient_id=_session_patient_id(session_id),
            ttl_seconds=0,
            release_hold_id=hold_id)
        _update_session_flag(session_id, "held_slot", None)
        _update_session_flag(session_id, "hold_id", None)
        next_text = (
            "No problem. Want me to look at other times, or "
            "should I drop this for now?"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      next_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "decline_hold",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=next_text,
            attach_greeting=False,
            disposition="held_declined")

    if confirmation != "confirm":
        # Ask for clarification.
        clarify_text = (
            "Did you want me to book that, or look at other "
            "options? A simple 'yes' or 'no' is fine."
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      clarify_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "clarify_confirmation",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=clarify_text,
            attach_greeting=False,
            disposition="awaiting_confirmation")

    # Invoke slot_book.
    book_start = datetime.now(timezone.utc)
    book_result = slot_book_tool(
        hold_id=hold_id,
        slot_id=held_slot["slot_id"],
        patient_id=_session_patient_id(session_id),
        notes=session.get("search_parameters", {}).get(
            "reason_for_visit"))
    book_latency_ms = int(
        (datetime.now(timezone.utc) - book_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="slot_book",
        arguments={
            "hold_id": hold_id,
            "slot_id": held_slot["slot_id"],
        },
        result_summary={
            "confirmation_id":
                book_result.get("confirmation_id"),
            "outcome": book_result.get("outcome"),
        },
        latency_ms=book_latency_ms,
        outcome=book_result.get("outcome", "unknown"))

    if book_result.get("outcome") != "booked":
        return _handle_booking_failure(
            session_id=session_id,
            channel=channel,
            failure=book_result,
            language=language)

    # Booking succeeded. Update session counters.
    _update_session_flag(
        session_id, "bookings_completed",
        int(session.get("bookings_completed", 0)) + 1)
    _update_session_flag(session_id, "held_slot", None)
    _update_session_flag(session_id, "hold_id", None)

    # Write the booking-event journal record.
    _write_booking_journal({
        "confirmation_id": book_result["confirmation_id"],
        "patient_id":      _session_patient_id(session_id),
        "slot_id":         held_slot["slot_id"],
        "provider":        held_slot.get("provider"),
        "location":        held_slot.get("location_name"),
        "visit_type":      held_slot.get("visit_type"),
        "appointment_time": held_slot.get("appointment_time"),
        "booking_initiated_through": "scheduling_bot",
        "session_id":      session_id,
        "booked_at":       _now_iso(),
        "event_type":      "booked",
    })

    _emit_event("booking_confirmed", {
        "session_id":      session_id,
        "confirmation_id": book_result["confirmation_id"],
        "patient_id":      _session_patient_id(session_id),
        "visit_type":      held_slot.get("visit_type"),
        "channel":         channel,
    })

    _put_metric("BookingCompleted", 1, {
        "channel":    channel,
        "visit_type": held_slot.get("visit_type"),
        "language":   language,
    })

    # Render confirmation. The text the user sees here is
    # checked by output screening (Step 9) against the actual
    # booking-tool result before delivery.
    confirmation_text = (
        f"Booked. Your confirmation is "
        f"{book_result['confirmation_id']}. You'll get a "
        f"confirmation text and email shortly. We'll also "
        f"send a reminder 48 hours before. Anything else I "
        f"can help with?"
    )

    _append_turn(session_id, {
        "speaker":   "assistant",
        "text":      confirmation_text,
        "timestamp": _now_iso(),
        "language":  language,
        "scheduling_action":  "booking_confirmed",
        "confirmation_id":
            book_result["confirmation_id"],
    })

    return _build_chat_reply(
        session_id=session_id,
        response_text=confirmation_text,
        attach_greeting=False,
        disposition="booked")

def _classify_confirmation_response(user_message: str) -> str:
    """Return one of: 'confirm', 'decline', 'modify', 'unclear'."""
    lowered = user_message.lower().strip()
    confirm_cues = ["yes", "yeah", "yep", "yup", "sure",
                     "book it", "book that", "ok", "okay",
                     "confirm", "please book"]
    decline_cues = ["no", "nope", "cancel", "never mind",
                     "nevermind", "forget it"]

    if any(cue == lowered or lowered.startswith(cue + " ")
           for cue in confirm_cues):
        return "confirm"
    if any(cue == lowered or lowered.startswith(cue + " ")
           for cue in decline_cues):
        return "decline"
    return "unclear"

def _write_booking_journal(record: dict) -> None:
    """
    Write a durable booking-event-journal record. The journal is
    the institution's audit-grade record of every booking,
    reschedule, and cancellation the bot performed, separate from
    the EHR's own record. In production the bucket has Object
    Lock in compliance mode and a retention period sized to the
    longest of HIPAA's six-year minimum, state-specific
    medical-records-retention rules, state-specific consumer-privacy-law
    retention rules where applicable, per-channel retention obligations,
    and the institutional regulatory floor.
    """
    key = (
        f"{INSTITUTION_ID}/"
        f"{datetime.now(timezone.utc):%Y/%m/%d}/"
        f"{record['confirmation_id']}.json")
    try:
        s3_client.put_object(
            Bucket=BOOKING_EVENT_JOURNAL_BUCKET,
            Key=key,
            Body=(json.dumps(record) + "\n").encode("utf-8"),
            ContentType="application/json",
            ServerSideEncryption="aws:kms")
    except Exception as exc:
        logger.error(
            "Booking-journal write failed for %s: %s",
            record.get("confirmation_id"), exc)