Recipe 11.3: 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 specific
# medications, doses, or symptoms. 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,
# never specific medication names or dose values. The full transcripts
# and full tool calls live in the audit pipeline (Firehose plus
# 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, Comprehend Medical, and
# Secrets Manager. The refill-bot response window is tight: the
# patient is staring at the chat widget waiting on a medication
# action. A retry storm that adds 5 seconds 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)
comprehend_medical    = boto3.client("comprehendmedical",
                                     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      = "refill-bot-conversation-state"
CONVERSATION_METADATA_TABLE   = "refill-bot-conversation-metadata"
TOOL_CALL_LEDGER_TABLE        = "refill-bot-tool-call-ledger"
COSIGNATURE_QUEUE_TABLE       = "refill-bot-cosignature-queue"
REFILL_EVENT_BUS_NAME         = "refill-bot-events-bus"
AUDIT_ARCHIVE_FIREHOSE_NAME   = "refill-bot-audit-archive"
REFILL_EVENT_JOURNAL_BUCKET   = "refill-bot-event-journal"
CLOUDWATCH_NAMESPACE          = "RefillBot"

# Bedrock Knowledge Base ID for medication-information corpus and
# patient-facing protocol phrasings.
KNOWLEDGE_BASE_ID             = "KB_PLACEHOLDER_ID"

# Bedrock Guardrail for restricted-topic filtering. Configure in
# the Bedrock console with restricted topics for clinical-advice,
# dose-change-recommendation, medication-discontinuation-guidance,
# and controlled-substance-auto-approval 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),
    ("COSIGNATURE_QUEUE_TABLE",      COSIGNATURE_QUEUE_TABLE),
    ("REFILL_EVENT_BUS_NAME",        REFILL_EVENT_BUS_NAME),
    ("AUDIT_ARCHIVE_FIREHOSE_NAME",  AUDIT_ARCHIVE_FIREHOSE_NAME),
    ("REFILL_EVENT_JOURNAL_BUCKET",  REFILL_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, and
# (most importantly) the active refill-protocol version. This is
# how a future audit reconstructs which protocol authorized a
# particular refill action.
PROMPT_VERSION                = "refill-bot-prompt-v3.1"
AGENT_VERSION                 = "refill-agent-v2.4"
ACTIVE_PROTOCOL_VERSION       = "refill-protocol-v4.2"
INSTITUTION_ID                = "riverside-clinic"
INSTITUTION_DISPLAY_NAME      = "Riverside Clinic"

# --- Model IDs ---
# Two model roles. Intent classification and medication-resolution
# disambiguation are cheap per-turn tasks where a smaller model
# earns its keep. The orchestration model decides which tool to
# call when, with what arguments; that work benefits from a
# stronger model with strong tool-use support.
#
# 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 medication resolution, ask the patient
# to clarify rather than acting on the wrong medication.
MED_RESOLUTION_CONFIDENCE_THRESHOLD = Decimal("0.80")

# Co-signature SLA. Most institutions choose 24-72 hours for
# asynchronous prescriber co-signature on auto-approved refills.
COSIGNATURE_SLA_HOURS           = 48

# Many institutions require authenticated sessions for any refill
# transmission. Set this to False to allow unauthenticated paths
# for refill actions (most institutions do not).
REQUIRE_AUTHENTICATED_FOR_REFILL = True

# --- Intents ---
REFILL_INTENTS = [
    "request_refill",
    "check_refill_status",
    "cancel_refill_request",
    "medication_question",
]

# Mapping of out-of-scope intents to handoff targets.
OUT_OF_SCOPE_INTENTS = {
    "clinical_question":     "nurse_triage",
    "medication_change":     "nurse_triage",
    "scheduling_request":    "scheduling_bot",
    "general_question":      "faq_bot",
    "out_of_scope":          "live_agent",
}

# --- Identity Verification Policy ---
# Maps (intent, authenticated_session) to a required assurance
# level. Refills generally have a higher floor than scheduling.
# 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 plus DOB plus one confirmation factor
#                    (last 4 of phone, ZIP code, or one-time code).
#                    Generally insufficient for refill actions
#                    when REQUIRE_AUTHENTICATED_FOR_REFILL is True.
#   "step_up":       basic plus a one-time code to a verified
#                    channel; used for higher-risk actions like
#                    early refills or controlled-substance routing.
IDENTITY_POLICY = {
    "request_refill": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "step_up"),
    ],
    "cancel_refill_request": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "step_up"),
    ],
    "check_refill_status": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "basic"),
    ],
    "medication_question": [
        (lambda ctx: ctx.get("authenticated"), "authenticated"),
        (lambda ctx: True, "basic"),
    ],
}

ASSURANCE_MATCH_THRESHOLDS = {
    "authenticated": Decimal("0.0"),  # already authenticated
    "basic":         Decimal("0.85"),
    "step_up":       Decimal("0.95"),
}

# --- Refill Protocol (illustrative) ---
# In production this is encoded as a versioned policy artifact
# reviewed by clinical leadership, not as a Python dict. Each
# medication class has explicit auto-approval criteria, monitoring
# requirements, dosing rules, and exceptions. The dict below is a
# starter for the demo only. The protocol cleanup is the single
# highest-leverage operational investment for the bot's quality.
REFILL_PROTOCOL = {
    "metformin": {
        "class":              "biguanide",
        "auto_approvable":    True,
        "monitoring_required": "a1c_within_365_days",
        "monitoring_max_days": 365,
        "max_days_supply":    90,
        "early_refill_threshold_days": 7,
    },
    "lisinopril": {
        "class":              "ace_inhibitor",
        "auto_approvable":    True,
        "monitoring_required": "creatinine_within_365_days",
        "monitoring_max_days": 365,
        "max_days_supply":    90,
        "early_refill_threshold_days": 7,
    },
    "atorvastatin": {
        "class":              "statin",
        "auto_approvable":    True,
        "monitoring_required": "lipid_panel_within_365_days",
        "monitoring_max_days": 365,
        "max_days_supply":    90,
        "early_refill_threshold_days": 7,
    },
    "levothyroxine": {
        "class":              "thyroid_hormone",
        "auto_approvable":    True,
        "monitoring_required": "tsh_within_365_days",
        "monitoring_max_days": 365,
        "max_days_supply":    90,
        "early_refill_threshold_days": 7,
    },
    "amiodarone": {
        "class":              "antiarrhythmic",
        "auto_approvable":    False,
        "specialist_only":    True,
        "specialty":          "cardiology",
        "routing_reason":
            "specialist_managed_amiodarone",
    },
    "methotrexate": {
        "class":              "dmard",
        "auto_approvable":    False,
        "specialist_only":    True,
        "specialty":          "rheumatology",
        "routing_reason":
            "specialist_managed_methotrexate",
    },
    # Controlled substances. The schedule is what matters; the
    # protocol routes regardless of any other rule.
    "oxycodone": {
        "class":               "opioid",
        "controlled_substance_schedule": "II",
        "auto_approvable":     False,
        "routing_reason":
            "controlled_substance_always_route",
    },
    "alprazolam": {
        "class":               "benzodiazepine",
        "controlled_substance_schedule": "IV",
        "auto_approvable":     False,
        "routing_reason":
            "controlled_substance_always_route",
    },
}

# --- High-Acuity Clinical Cues ---
# When detected in the user's message, route to the triage path
# rather than processing as a refill. Production uses Comprehend
# Medical's clinical-entity detection plus a tuned acuity classifier
# plus an explicit misuse-and-overdose detector; the keyword backstop
# below is illustrative.
HIGH_ACUITY_CUES = [
    "chest pain",
    "trouble breathing",
    "shortness of breath",
    "passed out",
    "lost consciousness",
    "suicidal",
    "want to hurt myself",
    "want to end my life",
    # Refill-specific misuse and overdose signals; these are the
    # signals where a refill conversation is masking a clinical
    # emergency.
    "took double",
    "took an extra",
    "overdose",
    "took too much",
    "ran out because i was taking more",
]

# --- 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."
    ),
    "medication_change": (
        "Starting, stopping, or changing a medication is "
        "something our clinical team handles, not me. Let "
        "me get a message to the nurse line for you, or "
        "you can call us at 555-0100."
    ),
    "scheduling_request": (
        "Booking, rescheduling, or canceling appointments "
        "goes through our scheduling assistant. Let me "
        "connect you with them."
    ),
    "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 refill assistant. "
    "I can help you request refills on medications you "
    "already take, check on requests you've already sent, "
    "and answer common medication questions. I can't change "
    "your medications or doses, can't start new prescriptions, "
    "and for clinical questions or anything urgent I'll "
    "connect you with the nursing team. What can I help with?"
)

CRISIS_ROUTE_TEMPLATE = (
    "If this is a medical emergency, please call 911 right "
    "now. For urgent symptoms or concerns about how you've "
    "been taking a medication, our nurse advice line at "
    "555-0100 can help right away. I'm a chatbot, so I "
    "can't safely handle an urgent situation. Let me "
    "connect you with a person who can."
)

REQUIRE_PORTAL_LOGIN_TEMPLATE = (
    "For a refill, I'll need you to be logged into the "
    "patient portal so I can verify it's you and look at "
    "your medication list. You can sign in at "
    "portal.example.org and come right back, or if you'd "
    "rather have a person handle it, our pharmacy line is "
    "at 555-0175."
)

NO_ACTIVE_MEDICATIONS_TEMPLATE = (
    "I'm not seeing any active medications on your record, "
    "which is unusual. Let me get this to our nursing team "
    "so they can sort out your medication list with you."
)

NO_MATCH_MEDICATION_TEMPLATE = (
    "I'm not finding that medication on your active "
    "medication list. It's possible it was prescribed "
    "outside our practice, was discontinued, or might be "
    "named slightly differently. Let me get this to our "
    "nursing team so they can sort it out with you."
)

DISCONTINUED_MEDICATION_TEMPLATE = (
    "I see {medication} on your record but it's marked as "
    "discontinued. Before I do anything, I want our "
    "nursing team to talk with you about it. Can I send "
    "them a note?"
)

SPECIALIST_MEDICATION_TEMPLATE = (
    "{medication} is managed by your {specialty} team, not "
    "primary care, so I can't refill it from here. The "
    "fastest path is to contact their office directly. "
    "Want me to send a message on your behalf?"
)

CONTROLLED_SUBSTANCE_ROUTING_TEMPLATE = (
    "Refills for {medication} need a clinician's review, so "
    "I can't send this one through automatically. I'm "
    "passing this to our clinical team and they'll reach "
    "out to you. If you're running low and feel you need "
    "it sooner, please call 555-0100."
)

EARLY_REFILL_ROUTING_TEMPLATE = (
    "It looks like you're asking for a refill on "
    "{medication} earlier than expected. I'm sending this "
    "to our clinical team to take a look. They'll reach "
    "out to you."
)

MONITORING_DUE_ROUTING_TEMPLATE = (
    "Your most recent {monitoring_label} on file is from "
    "{lab_date}, and the protocol on {medication} wants a "
    "more recent one before the next refill. I'm sending "
    "this to our nursing team. They can either find a "
    "result we don't have on file or get the lab scheduled."
)

REFILL_CONFIRM_FAILED_TEMPLATE = (
    "Something went wrong while I was trying to confirm "
    "that refill. I want to be straight with you: I'm not "
    "sure whether the refill went through. Let me get you "
    "to our pharmacy team at 555-0175 so we can verify "
    "and not have you waiting on a medication that didn't "
    "actually get sent."
)

PHARMACY_UNREACHABLE_TEMPLATE = (
    "Your pharmacy isn't responding right now, so the "
    "refill hasn't been transmitted yet. I've queued it "
    "for retry and our pharmacy team will follow up with "
    "you if it doesn't go through soon. You can also call "
    "the pharmacy directly to check."
)

TRANSMISSION_ERROR_TEMPLATE = (
    "I ran into a transmission error sending the refill. "
    "I've queued it for retry and noted it for our pharmacy "
    "team to follow up. If you need the medication soon, "
    "please call us at 555-0175."
)

INJECTION_REFUSAL_TEMPLATE = (
    "I can only help with medication refills here. Is "
    "there a refill 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."
)

NO_PENDING_REQUEST_TEMPLATE = (
    "I'm not finding any pending refill requests in your "
    "record. Want me to start a new one, or check on a "
    "specific medication?"
)

CANCEL_CONFIRMED_TEMPLATE = (
    "Done. I've cancelled that pending refill request. "
    "Anything else I can help with?"
)

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

CLINICAL_QUESTION_HANDOFF_TEMPLATE = (
    "That sounds like a clinical question I shouldn't try "
    "to answer. I'm sending it to our nursing team and "
    "they'll get back to you."
)

# --- 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),
}
# TODO (TechWriter): Code review W2 (WARNING). Same root cause as
# 11.01 / 11.02: the `account_long` 9-16-digit pattern matches any
# standalone digit run, including a phone number the bot just asked
# the patient to provide for identity verification. The contradictory
# loop is masked in 11.03 because REQUIRE_AUTHENTICATED_FOR_REFILL
# gates refill-action intents to authenticated sessions, but a
# deployment that flips that flag for status-check or
# medication-question intents over an unauthenticated channel hits
# the same loop documented in 11.02. Phase-gate the `account_long`
# check on the conversation phase (skip when the last assistant turn
# was an `ask_for_identifiers` / `ask_for_phone` / `step_up_requested`
# action), or tighten the pattern to require an account-context cue
# (e.g., (?:account|member|insurance)) within a small window. While
# editing, also extend `_redact_pii_for_logging` to cover ISO and
# slash-formatted dates of birth so the audit pipeline does not carry
# the patient's DOB in plain text after redaction.

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 _get_cosign_sla_tier(medication_class: str) -> str:
    """
    Return the co-signature SLA tier for a medication class.
    Tighter SLA tiers for higher-risk classes.
    """
    high_priority_classes = [
        "anticoagulant", "antiarrhythmic", "immunosuppressant"]
    if medication_class in high_priority_classes:
        return "urgent_24h"
    return "standard_48h"


def _get_escalation_policy(
        prescriber_id: str, medication_class: str) -> str:
    """
    Return the escalation policy identifier for a prescriber
    and medication class combination.
    """
    # In production this is looked up from a governance table.
    return f"escalation-{prescriber_id}-{medication_class}"

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":       "refill_bot",
            "DetailType":   detail_type,
            "Detail":       json.dumps(detail),
            "EventBusName": REFILL_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 refill 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.
    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)
    # TODO (TechWriter): Code review N2 (NOTE). The
    # read-modify-write above is a tutorial-grade race condition.
    # Two messages from the same channel arriving close together
    # both read message_count = N, both increment to N + 1, and the
    # last write wins, so message_count ends at N + 1 instead of
    # N + 2. The DynamoDB-native pattern is `update_item` with
    # UpdateExpression "SET #la = :ts ADD #mc :one"; switch to that
    # in production (and update the demo's mock to support
    # multi-attribute UpdateExpression and ADD; see N3).

    # 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,
        "active_protocol_version": ACTIVE_PROTOCOL_VERSION,
        # Conversation-level state filled in as we go.
        "intent":                  None,
        "extracted_parameters":    {},
        "active_medications":      [],
        "resolved_medication":     None,
        "protocol_decision":       None,
        "patient_pharmacies":      [],
        "patient_stated_context":  {},
        "crisis_detected":         False,
        "scope_violation_count":   0,
        "refills_auto_approved":   0,
        "refills_routed":          0,
        "refills_denied":          0,
        "refills_failed":          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 (with refill-
    specific overdose and misuse signals), prompt injection
    detection, and PHI minimization.
    """
    # Crisis detection. The refill bot does not handle the crisis
    # itself; it routes to triage. Refill-context misuse and
    # overdose signals (e.g., "I took double") are highest priority.
    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."""
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    try:
        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."""
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    if hasattr(table, "_find_session_key_by_session_id"):
        return table._find_session_key_by_session_id(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 refill flow.
    """
    classification = _classify_refill_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 request a refill, check on a "
            "request you've already sent, cancel a pending "
            "request, or ask a question about a medication?"
        )
        _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 refill
    # flow (Step 3 onwards).
    if intent in REFILL_INTENTS:
        _update_session_flag(session_id, "intent", intent)
        _update_session_flag(
            session_id,
            "extracted_parameters",
            classification.get("extracted_parameters", {}))
        return _route_to_refill_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 "
            "request a refill, check on one, cancel a "
            "request, or ask about a medication?"),
        attach_greeting=attach_greeting,
        disposition="clarification_requested")

def _classify_refill_intent(session_id: str,
                             user_message: str,
                             language: str) -> dict:
    """
    Classify the user's message into a refill intent and extract
    any preliminary parameters (medication descriptor, pharmacy
    preference, patient-stated context).
    """
    in_scope_csv = ", ".join(REFILL_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 "
        "prescription-refill 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'
        '    "medication_descriptor": "<patient'
        "'s text describing the medication, or null>\",\n"
        '    "pharmacy_descriptor": "<pharmacy reference '
        'or null>",\n'
        '    "patient_stated_context": "<any context the '
        'patient volunteered, or null>",\n'
        '    "question": "<for medication_question: the '
        'question text, 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, side-effect "
        "concerns, should-I-take, dose questions, "
        "interaction worries) is ALWAYS clinical_question.\n"
        "- A request to START, STOP, CHANGE, or TITRATE a "
        "medication is ALWAYS medication_change.\n"
        "- 'I want to stop taking X' or 'I want to come "
        "off X' is medication_change, NOT request_refill.\n"
        "- 'Can I get my X refilled' is request_refill.\n"
        "- 'When will my refill be ready' is "
        "check_refill_status.\n"
        "- 'How does this medication interact with...' is "
        "medication_question.\n"
        "- A request to schedule, reschedule, or cancel an "
        "APPOINTMENT is scheduling_request.\n"
        "- General clinic questions (parking, hours, "
        "directions) are general_question."
    )
    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_refill_flow(session_id: str,
                           channel: str,
                           user_message: str,
                           intent: str,
                           extracted_parameters: dict,
                           auth_context: dict,
                           attach_greeting: bool,
                           language: str) -> dict:
    """
    The refill-specific flow needs identity verification before
    any action that touches the patient's medication record.
    """
    # Determine the required assurance level for this (intent,
    # auth state).
    policy_ctx = {
        "authenticated": auth_context.get("authenticated", False),
    }
    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)

    # Many institutions require authenticated sessions for any
    # refill action that results in transmission. Status-check
    # intents may be allowed at lower assurance.
    if (REQUIRE_AUTHENTICATED_FOR_REFILL
            and intent in ("request_refill",
                           "cancel_refill_request")
            and not auth_context.get("authenticated")):
        _put_metric("PortalLoginRequired", 1, {
            "channel":  channel,
            "intent":   intent,
            "language": language,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      REQUIRE_PORTAL_LOGIN_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "identity_action": "require_portal_login",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=REQUIRE_PORTAL_LOGIN_TEMPLATE,
            attach_greeting=attach_greeting,
            disposition="require_portal_login")

    # Otherwise, gather identifiers conversationally for the
    # status-check-only intents that we still allow over an
    # unauthenticated channel.
    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. The name or date of birth might be "
            "different from what we have on file. I can also "
            "connect you with our pharmacy team at 555-0175."
        )
        _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)."""
    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
    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

    conf_match = re.search(
        r"(?<!\d)(\d{4})(?!\d)", combined)
    confirmation = conf_match.group(1) if conf_match else None
    # TODO (TechWriter): Code review W1 (WARNING). Same root cause as
    # 11.01 / 11.02: this regex returns the FIRST standalone 4-digit
    # sequence, which on the canonical "Marcus Chen, 1979-03-14, 7842"
    # input picks "1979" (the DOB year) instead of "7842" (the
    # last-four-of-phone the bot's prompt asked for). The bug is masked
    # in 11.03 because REQUIRE_AUTHENTICATED_FOR_REFILL gates the
    # refill-action intents to authenticated sessions, but the bug fires
    # immediately if a deployment turns that flag off. Use re.findall
    # and pick the LAST match, ideally as a shared helper across the
    # 11.01-11.04 recipes so the fix lands once.

    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.

    request_refill goes to medication resolution; check_refill_status
    and cancel_refill_request go through their own simpler paths;
    medication_question retrieves curated content.
    """
    if intent == "request_refill":
        return _resolve_medication(
            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 == "check_refill_status":
        return _handle_status_check(
            session_id=session_id,
            channel=channel,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language)

    if intent == "cancel_refill_request":
        return _handle_cancel_request(
            session_id=session_id,
            channel=channel,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language)

    if intent == "medication_question":
        return _handle_medication_question(
            session_id=session_id,
            channel=channel,
            extracted_parameters=extracted_parameters,
            attach_greeting=attach_greeting,
            language=language)

    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 _resolve_medication(session_id: str,
                         channel: str,
                         extracted_parameters: dict,
                         attach_greeting: bool,
                         language: str,
                         verified_just_now: bool,
                         patient_first_name: Optional[str]
                         ) -> dict:
    """Pull medication list, resolve the descriptor, run preflight checks."""
    # Step 4A: pull the structured medication list from FHIR.
    list_start = datetime.now(timezone.utc)
    med_list_result = medication_list_lookup_tool(
        patient_id=_session_patient_id(session_id),
        active_only=True)
    list_latency_ms = int(
        (datetime.now(timezone.utc) - list_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="medication_list_lookup",
        arguments={"active_only": True},
        result_summary={
            "medication_count":
                len(med_list_result.get("medications", [])),
        },
        latency_ms=list_latency_ms,
        outcome="ok"
            if med_list_result.get("medications")
            else "no_active_medications")

    medications = med_list_result.get("medications", [])
    _update_session_flag(
        session_id, "active_medications", medications)

    # Stash patient-level chart context the protocol step will need.
    _update_session_flag(
        session_id, "patient_pharmacies",
        med_list_result.get("pharmacies", []))
    _update_session_flag(
        session_id, "patient_allergies",
        med_list_result.get("allergies", []))
    _update_session_flag(
        session_id, "patient_conditions",
        med_list_result.get("conditions", []))

    if not medications:
        # No active medications on file. Route to nursing for
        # chart reconciliation.
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      NO_ACTIVE_MEDICATIONS_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action":
                "no_active_medications_routed",
        })
        _emit_event("refill_routed", {
            "session_id": session_id,
            "disposition":
                "no_active_medications_on_file",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=NO_ACTIVE_MEDICATIONS_TEMPLATE,
            attach_greeting=attach_greeting,
            disposition="no_active_medications")

    # Step 4B: resolve the patient's descriptor against the
    # active list.
    descriptor = extracted_parameters.get(
        "medication_descriptor", "")
    resolution_start = datetime.now(timezone.utc)
    resolution_result = medication_resolution_tool(
        patient_descriptor=descriptor,
        medication_list=medications,
        language=language)
    resolution_latency_ms = int(
        (datetime.now(timezone.utc) - resolution_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="medication_resolution",
        arguments={"descriptor": descriptor},
        result_summary={
            "resolution_status":
                resolution_result.get("status"),
            "resolved_med_id":
                resolution_result.get("medication", {}).get("id"),
            "confidence":
                str(resolution_result.get("confidence", 0)),
        },
        latency_ms=resolution_latency_ms,
        outcome=resolution_result.get("status", "unknown"))

    status = resolution_result.get("status")

    # Step 4C: handle resolution outcomes.
    if status == "ambiguous":
        candidates = resolution_result.get("candidates", [])
        candidate_text = "\n".join(
            f"- {c.get('display_name')}"
            for c in candidates[:4])
        ask_text = (
            "A few of your medications could fit what you "
            f"described. Which one did you mean?\n{candidate_text}"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      ask_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "clarify_medication",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=ask_text,
            attach_greeting=attach_greeting,
            disposition="clarify_medication")

    if status == "no_match":
        # The patient mentioned a medication that is not on
        # their active list. Could be discontinued, an outside
        # prescriber's, a misremembering, or a naming confusion.
        # The bot does not guess; it routes for clinical follow-up.
        _emit_event("refill_routed", {
            "session_id": session_id,
            "disposition":
                "medication_not_on_active_list",
            "descriptor": descriptor,
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      NO_MATCH_MEDICATION_TEMPLATE,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action":
                "medication_not_on_list_routed",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=NO_MATCH_MEDICATION_TEMPLATE,
            attach_greeting=attach_greeting,
            disposition="no_match_medication")

    if status == "discontinued_match":
        # Medication exists on the record but is marked
        # discontinued. Route for reconciliation rather than
        # auto-acting.
        med = resolution_result.get("medication", {})
        text = DISCONTINUED_MEDICATION_TEMPLATE.format(
            medication=med.get("display_name", "that medication"))
        _emit_event("refill_routed", {
            "session_id": session_id,
            "disposition": "discontinued_medication",
            "medication_id": med.get("id"),
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action":
                "discontinued_medication_routed",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=text,
            attach_greeting=attach_greeting,
            disposition="discontinued_medication_routed")

    confidence = Decimal(str(
        resolution_result.get("confidence", 0)))
    if confidence < MED_RESOLUTION_CONFIDENCE_THRESHOLD:
        # Low-confidence resolution. Ask the patient to confirm
        # before acting.
        med = resolution_result.get("medication", {})
        confirm_text = (
            f"Just to make sure I have the right medication: "
            f"is this {med.get('display_name', 'medication')}?"
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      confirm_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "confirm_medication",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=confirm_text,
            attach_greeting=attach_greeting,
            disposition="confirm_medication")

    medication = resolution_result.get("medication", {})

    # Step 4D: prescriber-authority check. Specialist medications
    # do not refill from primary care; surface the boundary
    # transparently and route.
    protocol_entry = REFILL_PROTOCOL.get(
        medication.get("name", "").lower(), {})
    if protocol_entry.get("specialist_only"):
        text = SPECIALIST_MEDICATION_TEMPLATE.format(
            medication=medication.get(
                "display_name", "that medication"),
            specialty=protocol_entry.get("specialty",
                                         "specialty"))
        _emit_event("refill_routed", {
            "session_id": session_id,
            "disposition": "specialist_managed_medication",
            "medication_id": medication.get("id"),
            "specialty": protocol_entry.get("specialty"),
        })
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "specialist_routed",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=text,
            attach_greeting=attach_greeting,
            disposition="specialist_routed")

    # Resolved successfully. Stash and continue to protocol
    # evaluation.
    _update_session_flag(
        session_id, "resolved_medication", medication)

    if extracted_parameters.get("patient_stated_context"):
        _update_session_flag(
            session_id, "patient_stated_context", {
                "raw":
                    extracted_parameters[
                        "patient_stated_context"],
            })

    return _evaluate_protocol(
        session_id=session_id,
        channel=channel,
        attach_greeting=attach_greeting,
        language=language,
        verified_just_now=verified_just_now,
        patient_first_name=patient_first_name)

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 _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 _evaluate_protocol(session_id: str,
                        channel: str,
                        attach_greeting: bool,
                        language: str,
                        verified_just_now: bool,
                        patient_first_name: Optional[str]
                        ) -> dict:
    """Reconcile labs, screen interactions, evaluate the protocol."""
    session = _session_state(session_id)
    medication = session.get("resolved_medication") or {}

    # Step 5A: lab reconciliation. Check for the relevant lab
    # before running the protocol so a recent outside-lab result
    # does not get missed (the Eleanor failure mode).
    lab_recon_start = datetime.now(timezone.utc)
    lab_recon_result = lab_reconciliation_tool(
        patient_id=_session_patient_id(session_id),
        medication_class=medication.get("class"),
        lookback_days=365)
    lab_recon_latency_ms = int(
        (datetime.now(timezone.utc) - lab_recon_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="lab_reconciliation",
        arguments={
            "medication_class": medication.get("class"),
            "lookback_days": 365,
        },
        result_summary={
            "relevant_lab_count":
                len(lab_recon_result.get(
                    "relevant_labs", [])),
            "most_recent_lab_date":
                lab_recon_result.get("most_recent_date"),
        },
        latency_ms=lab_recon_latency_ms,
        outcome="ok")

    # Step 5B: drug-interaction screening.
    interaction_start = datetime.now(timezone.utc)
    interaction_result = interaction_screening_tool(
        patient_id=_session_patient_id(session_id),
        medication=medication,
        active_medications=session.get(
            "active_medications", []),
        allergies=session.get("patient_allergies", []),
        conditions=session.get("patient_conditions", []))
    interaction_latency_ms = int(
        (datetime.now(timezone.utc) - interaction_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="interaction_screening",
        arguments={"medication_id": medication.get("id")},
        result_summary={
            "interactions_found":
                len(interaction_result.get(
                    "interactions", [])),
            "max_severity":
                interaction_result.get("max_severity"),
        },
        latency_ms=interaction_latency_ms,
        outcome="ok")

    # Step 5C: protocol evaluation.
    protocol_start = datetime.now(timezone.utc)
    protocol_result = protocol_evaluate_tool(
        patient_id=_session_patient_id(session_id),
        medication=medication,
        chart_context={
            "active_medications":
                session.get("active_medications", []),
            "allergies":
                session.get("patient_allergies", []),
            "conditions":
                session.get("patient_conditions", []),
            "relevant_labs":
                lab_recon_result.get("relevant_labs", []),
            "interactions":
                interaction_result.get("interactions", []),
        },
        request_context={
            "days_since_last_fill":
                medication.get("days_since_last_fill"),
            "refills_remaining":
                medication.get("refills_remaining"),
            "patient_stated_context":
                session.get("patient_stated_context", {}),
        },
        protocol_version=ACTIVE_PROTOCOL_VERSION)
    protocol_latency_ms = int(
        (datetime.now(timezone.utc) - protocol_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="protocol_evaluate",
        arguments={
            "medication_id": medication.get("id"),
            "protocol_version": ACTIVE_PROTOCOL_VERSION,
        },
        result_summary={
            "disposition": protocol_result.get("disposition"),
            "rules_fired": protocol_result.get(
                "rules_fired", []),
            "protocol_version":
                protocol_result.get("protocol_version"),
        },
        latency_ms=protocol_latency_ms,
        outcome=protocol_result.get("disposition", "unknown"))

    # Step 5D: controlled-substance triple-defense. Regardless
    # of what the protocol_evaluate returned, force the safe
    # disposition for any controlled-substance schedule and
    # alarm if the protocol returned anything else.
    schedule = medication.get("controlled_substance_schedule")
    if schedule in ("II", "III", "IV", "V"):
        if (protocol_result.get("disposition")
                != "controlled_substance_always_route"):
            logger.warning(
                "Protocol misclassification on controlled "
                "substance for session %s: returned %s",
                session_id,
                protocol_result.get("disposition"))
            _put_metric(
                "ProtocolControlledSubstanceMisclassification",
                1, {
                    "channel":  channel,
                    "language": language,
                })
            protocol_result["disposition"] = (
                "controlled_substance_always_route")
            protocol_result.setdefault(
                "rules_fired", []).append(
                "controlled_substance_triple_defense")

    _update_session_flag(
        session_id, "protocol_decision", protocol_result)

    return _execute_disposition(
        session_id=session_id,
        channel=channel,
        attach_greeting=attach_greeting,
        language=language,
        verified_just_now=verified_just_now,
        patient_first_name=patient_first_name)

def _execute_disposition(session_id: str,
                          channel: str,
                          attach_greeting: bool,
                          language: str,
                          verified_just_now: bool,
                          patient_first_name: Optional[str]
                          ) -> dict:
    """Execute the protocol's disposition for the resolved medication."""
    session = _session_state(session_id)
    decision = session.get("protocol_decision") or {}
    medication = session.get("resolved_medication") or {}
    disposition = decision.get("disposition")

    if disposition == "auto_approve":
        return _execute_auto_approve(
            session_id=session_id,
            channel=channel,
            decision=decision,
            medication=medication,
            attach_greeting=attach_greeting,
            language=language,
            verified_just_now=verified_just_now,
            patient_first_name=patient_first_name)

    if disposition in ("route_to_prescriber",
                        "route_with_monitoring_due",
                        "route_with_clinical_question",
                        "controlled_substance_always_route",
                        "early_refill_route"):
        return _execute_clinical_routing(
            session_id=session_id,
            channel=channel,
            decision=decision,
            medication=medication,
            attach_greeting=attach_greeting,
            language=language)

    if disposition == "deny_with_reason":
        return _execute_denial(
            session_id=session_id,
            channel=channel,
            decision=decision,
            medication=medication,
            attach_greeting=attach_greeting,
            language=language)

    # Defensive default: anything unexpected routes to clinical.
    logger.warning(
        "Unexpected disposition %s; routing to clinical",
        disposition)
    return _execute_clinical_routing(
        session_id=session_id,
        channel=channel,
        decision=decision,
        medication=medication,
        attach_greeting=attach_greeting,
        language=language)

def _execute_auto_approve(session_id: str,
                            channel: str,
                            decision: dict,
                            medication: dict,
                            attach_greeting: bool,
                            language: str,
                            verified_just_now: bool,
                            patient_first_name: Optional[str]
                            ) -> dict:
    """E-prescribe, enqueue co-signature, journal the event."""
    session = _session_state(session_id)

    # Step 6A: select dispensing pharmacy.
    pharmacy = _select_dispensing_pharmacy(
        medication=medication,
        patient_pharmacies=session.get(
            "patient_pharmacies", []),
        patient_preference=None)

    if pharmacy is None:
        # No pharmacy on file or no clear default. Ask the
        # patient.
        ask_text = (
            "Where would you like me to send this? I see "
            "more than one pharmacy on your record."
        )
        _append_turn(session_id, {
            "speaker":   "assistant",
            "text":      ask_text,
            "timestamp": _now_iso(),
            "language":  language,
            "scheduling_action": "ask_for_pharmacy",
        })
        return _build_chat_reply(
            session_id=session_id,
            response_text=ask_text,
            attach_greeting=attach_greeting,
            disposition="ask_for_pharmacy")

    # Step 6B: e-prescribe.
    eprescribe_start = datetime.now(timezone.utc)
    eprescribe_result = e_prescribe_tool(
        patient_id=_session_patient_id(session_id),
        medication=medication,
        pharmacy=pharmacy,
        quantity=medication.get("standard_quantity"),
        days_supply=medication.get("standard_days_supply"),
        refills_authorized=medication.get(
            "standard_refills", 3),
        prescribing_provider_id=medication.get(
            "prescribing_provider_id"),
        authorization_basis={
            "bot_initiated":     True,
            "protocol_version":
                decision.get("protocol_version"),
            "rules_fired":
                decision.get("rules_fired", []),
        })
    eprescribe_latency_ms = int(
        (datetime.now(timezone.utc) - eprescribe_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="e_prescribe",
        arguments={
            "medication_id": medication.get("id"),
            "pharmacy_id":   pharmacy.get("id"),
        },
        result_summary={
            "outcome": eprescribe_result.get("outcome"),
            "prescription_id":
                eprescribe_result.get("prescription_id"),
        },
        latency_ms=eprescribe_latency_ms,
        outcome=eprescribe_result.get("outcome", "unknown"))

    if eprescribe_result.get("outcome") != "transmitted":
        return _handle_eprescribe_failure(
            session_id=session_id,
            channel=channel,
            failure=eprescribe_result,
            medication=medication,
            language=language)

    # Step 6C: enqueue prescriber co-signature.
    cosignature_queue.put_item(Item=_to_decimal({
        "prescription_id":
            eprescribe_result["prescription_id"],
        "prescriber_id":
            medication.get("prescribing_provider_id"),
        "patient_id":
            _session_patient_id(session_id),
        "medication_id":
            medication.get("id"),
        "protocol_version":
            decision.get("protocol_version"),
        "medication_class_protocol_version":
            decision.get("medication_class_protocol_version"),
        "delegation_version":
            decision.get("delegation_version"),
        "rules_fired":
            decision.get("rules_fired", []),
        "sla_tier":
            _get_cosign_sla_tier(medication.get("class")),
        "sla_deadline": (
            datetime.now(timezone.utc)
            + timedelta(hours=COSIGNATURE_SLA_HOURS)
        ).isoformat(),
        "escalation_policy":
            _get_escalation_policy(
                medication.get("prescribing_provider_id"),
                medication.get("class")),
        "session_id": session_id,
        "enqueued_at": _now_iso(),
        "status": "pending",
    }))

    _emit_event("cosignature_pending", {
        "session_id":      session_id,
        "prescription_id":
            eprescribe_result["prescription_id"],
        "prescriber_id":
            medication.get("prescribing_provider_id"),
    })

    # Step 6D: write the refill-event journal.
    # The journal carries only structural fields. Free-text
    # context (data_consulted, rules_fired details,
    # patient_stated_context) routes to the per-conversation
    # archive in production; here we write rule-IDs only.
    rules_fired_ids = [
        r.get("rule_id", r) if isinstance(r, dict) else r
        for r in decision.get("rules_fired", [])]
    _write_refill_journal({
        "event_type":      "refill_auto_approved",
        "event_id":         str(uuid.uuid4()),
        "patient_id":
            _session_patient_id(session_id),
        "medication_id":   medication.get("id"),
        "medication_name": medication.get("name"),
        "medication_strength":
            medication.get("strength"),
        "prescription_id":
            eprescribe_result["prescription_id"],
        "dispensing_pharmacy_id":
            pharmacy.get("id"),
        "prescribing_provider_id":
            medication.get("prescribing_provider_id"),
        "protocol_version":
            decision.get("protocol_version"),
        "medication_class_protocol_version":
            decision.get("medication_class_protocol_version"),
        "delegation_version":
            decision.get("delegation_version"),
        "rules_fired_summary":   rules_fired_ids,
        "data_consulted_archive_ref":
            f"{session_id}/data_consulted",
        "rules_fired_archive_ref":
            f"{session_id}/rules_fired",
        "patient_stated_context_archive_ref":
            f"{session_id}/patient_context",
        "session_id":      session_id,
        "initiated_at":    _now_iso(),
    })

    _emit_event("refill_auto_approved", {
        "session_id":      session_id,
        "prescription_id":
            eprescribe_result["prescription_id"],
        "patient_id":
            _session_patient_id(session_id),
        "medication_class": medication.get("class"),
        "channel":         channel,
    })

    _put_metric("RefillAutoApproved", 1, {
        "channel":         channel,
        "medication_class":
            medication.get("class", "unknown"),
        "language":        language,
    })

    # Update session counters.
    _update_session_flag(
        session_id, "refills_auto_approved",
        int(session.get("refills_auto_approved", 0)) + 1)

    # Step 6E: render confirmation. The output safety screen
    # (Step 9) verifies this against the actual e_prescribe
    # tool result before delivery.
    response_text = _build_approval_response(
        medication=medication,
        pharmacy=pharmacy,
        prescription_id=eprescribe_result["prescription_id"],
        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": "refill_auto_approved",
        "prescription_id":
            eprescribe_result["prescription_id"],
    })

    return _build_chat_reply(
        session_id=session_id,
        response_text=response_text,
        attach_greeting=attach_greeting,
        disposition="auto_approved")

def _execute_clinical_routing(session_id: str,
                                channel: str,
                                decision: dict,
                                medication: dict,
                                attach_greeting: bool,
                                language: str) -> dict:
    """Package a structured ticket and route to the right inbox."""
    session = _session_state(session_id)
    disposition = decision.get("disposition")

    routing_target = _routing_target_for(disposition)

    ticket = {
        "intent": "refill_request",
        "patient_id": _session_patient_id(session_id),
        "medication": medication,
        "protocol_decision": decision,
        "patient_stated_context":
            session.get("patient_stated_context", {}),
        "session_id": session_id,
    }

    routing_start = datetime.now(timezone.utc)
    routing_result = clinical_routing_tool(
        target=routing_target,
        ticket=ticket)
    routing_latency_ms = int(
        (datetime.now(timezone.utc) - routing_start)
        .total_seconds() * 1000)

    _audit_tool_call(
        session_id=session_id,
        tool="clinical_routing",
        arguments={
            "target": routing_target,
            "disposition": disposition,
        },
        result_summary={
            "outcome": routing_result.get("outcome"),
            "queue_position":
                routing_result.get("queue_position"),
            "estimated_sla":
                routing_result.get("estimated_sla"),
        },
        latency_ms=routing_latency_ms,
        outcome=routing_result.get("outcome", "unknown"))

    _write_refill_journal({
        "event_type":      "refill_routed_to_clinician",
        "event_id":         str(uuid.uuid4()),
        "patient_id":
            _session_patient_id(session_id),
        "medication_id":   medication.get("id"),
        "medication_name": medication.get("name"),
        "protocol_version":
            decision.get("protocol_version"),
        "disposition":     disposition,
        "rules_fired":
            decision.get("rules_fired", []),
        "routed_to":        routing_target,
        "session_id":      session_id,
        "initiated_at":    _now_iso(),
    })

    _emit_event("refill_routed", {
        "session_id":   session_id,
        "patient_id":
            _session_patient_id(session_id),
        "disposition": disposition,
        "routed_to":    routing_target,
        "channel":     channel,
    })

    _put_metric("RefillRouted", 1, {
        "channel":     channel,
        "disposition": disposition,
        "language":    language,
    })

    _update_session_flag(
        session_id, "refills_routed",
        int(session.get("refills_routed", 0)) + 1)

    response_text = _build_routing_response(
        medication=medication,
        decision=decision,
        routing_target=routing_target,
        estimated_sla=routing_result.get("estimated_sla"))

    _append_turn(session_id, {
        "speaker":   "assistant",
        "text":      response_text,
        "timestamp": _now_iso(),
        "language":  language,
        "scheduling_action": "refill_routed",
        "disposition": disposition,
    })

    return _build_chat_reply(
        session_id=session_id,
        response_text=response_text,
        attach_greeting=attach_greeting,
        disposition="routed")

def _execute_denial(session_id: str,
                     channel: str,
                     decision: dict,
                     medication: dict,
                     attach_greeting: bool,
                     language: str) -> dict:
    """Journal the denial and explain transparently."""
    session = _session_state(session_id)

    _write_refill_journal({
        "event_type":      "refill_denied",
        "event_id":         str(uuid.uuid4()),
        "patient_id":
            _session_patient_id(session_id),
        "medication_id":   medication.get("id"),
        "medication_name": medication.get("name"),
        "protocol_version":
            decision.get("protocol_version"),
        "rules_fired":
            decision.get("rules_fired", []),
        "denial_reason":
            decision.get("denial_reason"),
        "session_id":      session_id,
        "initiated_at":    _now_iso(),
    })

    _emit_event("refill_denied", {
        "session_id": session_id,
        "patient_id":
            _session_patient_id(session_id),
        "reason": decision.get("denial_reason"),
        "channel": channel,
    })

    _put_metric("RefillDenied", 1, {
        "channel":   channel,
        "reason":    decision.get("denial_reason", "unknown"),
        "language":  language,
    })

    _update_session_flag(
        session_id, "refills_denied",
        int(session.get("refills_denied", 0)) + 1)

    response_text = (
        "I'm not able to refill that one based on our "
        "protocol. The reason: "
        f"{decision.get('patient_facing_reason', 'a clinical rule that needs review')}. "
        "Our pharmacy team can help if you'd like to talk it "
        "through; their number is 555-0175."
    )

    _append_turn(session_id, {
        "speaker":   "assistant",
        "text":      response_text,
        "timestamp": _now_iso(),
        "language":  language,
        "scheduling_action": "refill_denied",
    })

    return _build_chat_reply(
        session_id=session_id,
        response_text=response_text,
        attach_greeting=attach_greeting,
        disposition="denied")

def _routing_target_for(disposition: str) -> str:
    """Map disposition to a clinical-inbox routing target."""
    return {
        "route_to_prescriber":             "prescriber_inbox",
        "route_with_monitoring_due":       "nurse_triage",
        "route_with_clinical_question":    "nurse_triage",
        "controlled_substance_always_route":
            "prescriber_inbox_controlled_substance",
        "early_refill_route":              "prescriber_inbox",
    }.get(disposition, "nurse_triage")

def _select_dispensing_pharmacy(medication: dict,
                                  patient_pharmacies: list,
                                  patient_preference: Optional[dict]
                                  ) -> Optional[dict]:
    """Pick the right pharmacy. Prefer the medication's last-fill location."""
    if patient_preference:
        return patient_preference
    last_fill_pharmacy_id = medication.get(
        "last_fill_pharmacy_id")
    for pharmacy in patient_pharmacies:
        if pharmacy.get("id") == last_fill_pharmacy_id:
            return pharmacy
    if len(patient_pharmacies) == 1:
        return patient_pharmacies[0]
    if patient_pharmacies:
        # No clear default. Returning None signals the bot to ask.
        return None
    return None

def _build_approval_response(medication: dict,
                              pharmacy: dict,
                              prescription_id: str,
                              verified_just_now: bool,
                              patient_first_name: Optional[str]
                              ) -> str:
    """Render the auto-approved-refill response."""
    lines = []
    if verified_just_now and patient_first_name:
        lines.append(f"Thanks {patient_first_name}.")
    lines.append(
        f"I've sent your "
        f"{medication.get('display_name', 'refill')} to "
        f"{pharmacy.get('display_name', 'your pharmacy')}. "
        f"Confirmation number is {prescription_id}. The "
        f"pharmacy will text you when it's ready, usually "
        f"later today. Anything else I can help with?"
    )
    return " ".join(lines)

def _build_routing_response(medication: dict,
                              decision: dict,
                              routing_target: str,
                              estimated_sla: Optional[str]
                              ) -> str:
    """Render the routing response, with a transparent reason."""
    disposition = decision.get("disposition")
    sla_text = estimated_sla or "the next 1-2 business days"

    if disposition == "controlled_substance_always_route":
        return CONTROLLED_SUBSTANCE_ROUTING_TEMPLATE.format(
            medication=medication.get(
                "display_name", "that medication"))

    if disposition == "early_refill_route":
        return EARLY_REFILL_ROUTING_TEMPLATE.format(
            medication=medication.get(
                "display_name", "that medication"))

    if disposition == "route_with_monitoring_due":
        rules = decision.get("rules_fired", [])
        # The protocol stamps which monitoring rule fired and
        # what label to surface.
        return MONITORING_DUE_ROUTING_TEMPLATE.format(
            medication=medication.get(
                "display_name", "that medication"),
            monitoring_label=decision.get(
                "monitoring_label", "lab"),
            lab_date=decision.get(
                "monitoring_last_date", "your last result"))

    # Generic routing message.
    return (
        f"I'm passing your refill request for "
        f"{medication.get('display_name', 'that medication')} "
        f"to our clinical team. They'll review and get back "
        f"to you within {sla_text}."
    )

def _write_refill_journal(record: dict) -> None:
    """
    Write a durable refill-event-journal record. The journal is
    the institution's audit-grade clinical-record-event log,
    separately governed from the conversation log. 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 PDMP retention rules where applicable for
    controlled-substance-related records, state-specific
    pharmacy-record retention rules, state-specific
    consumer-privacy-law retention rules (CCPA/CPRA, VCDPA,
    CPA, etc.), per-channel retention obligations (TCPA/10DLC
    for SMS), and the institutional regulatory floor.
    """
    key = (
        f"{INSTITUTION_ID}/"
        f"{datetime.now(timezone.utc):%Y/%m/%d}/"
        f"{record['event_id']}.json")
    try:
        s3_client.put_object(
            Bucket=REFILL_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(
            "Refill-journal write failed for %s: %s",
            record.get("event_id"), exc)