Recipe 4.5: Python Implementation Example

pip install boto3 pandas numpy

import json
import logging
import time
import uuid
import datetime
from datetime import timezone, timedelta
from decimal import Decimal

import boto3
from botocore.config import Config

# Structured logging. In production, ship JSON-formatted records to
# CloudWatch Logs Insights. Never log a raw (patient_id, therapeutic_class,
# barrier) join along with clinical context; the row implicitly identifies
# both the medication and the suspected reason. The barrier-classifications
# table and the recommendation log are highly inferential PHI.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from SageMaker, DynamoDB, Bedrock,
# Kinesis, S3, Athena, and SES during the weekly batch run. Adherence
# adds pharmacy-claim ingestion bursts on top of the recommendation
# load; both stages share the same retry envelope.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 5, "mode": "adaptive"})

# Module-level clients. Reused across Lambda invocations in warm containers.
sagemaker_client = boto3.client("sagemaker", config=BOTO3_RETRY_CONFIG)
sagemaker_runtime = boto3.client("sagemaker-runtime", config=BOTO3_RETRY_CONFIG)
sagemaker_featurestore = boto3.client(
    "sagemaker-featurestore-runtime", config=BOTO3_RETRY_CONFIG
)
dynamodb = boto3.resource("dynamodb", config=BOTO3_RETRY_CONFIG)
bedrock_runtime = boto3.client("bedrock-runtime", config=BOTO3_RETRY_CONFIG)
kinesis_client = boto3.client("kinesis", config=BOTO3_RETRY_CONFIG)
s3_client = boto3.client("s3", config=BOTO3_RETRY_CONFIG)
athena_client = boto3.client("athena", config=BOTO3_RETRY_CONFIG)
cloudwatch_client = boto3.client("cloudwatch", config=BOTO3_RETRY_CONFIG)
ses_client = boto3.client("ses", config=BOTO3_RETRY_CONFIG)

# --- Model Configuration ---
# Three distinct LLM use cases, each wired to a small/fast model.
# Haiku-class hits the cost target at scale; larger frontier models
# add cost without meaningfully better tailoring on these prompt
# shapes. Production picks the model per use case based on observed
# quality regressions, not on a single uniform default.
BARRIER_CLASSIFIER_MODEL_ID = "anthropic.claude-3-5-haiku-20241022-v1:0"
REMINDER_MODEL_ID           = "anthropic.claude-3-5-haiku-20241022-v1:0"
PHARMACIST_BRIEF_MODEL_ID   = "anthropic.claude-3-5-haiku-20241022-v1:0"
PCP_BRIEFING_MODEL_ID       = "anthropic.claude-3-5-haiku-20241022-v1:0"

# Names of the SageMaker model artifacts. The need scorer is per
# therapeutic class (the clinical risk of continued non-adherence
# differs enough between statins and oral diabetes meds that one
# model per class produces better calibration). Engagement and uplift
# are per intervention (a text reminder's engagement signature is
# not the same as a pharmacist-consult's signature).
NEED_MODEL_NAMES = {
    "statins":              "need-statins-v4",
    "ras_antagonists":      "need-ras-v3",
    "oral_diabetes":        "need-oral-diabetes-v4",
}
ENGAGEMENT_MODEL_NAMES = {
    "text_reminder":        "engagement-text-v3",
    "education":            "engagement-education-v2",
    "pharmacist_consult":   "engagement-pharmacist-v3",
    "cost_assistance":      "engagement-cost-assist-v2",
    "med_sync":             "engagement-med-sync-v1",
    "regimen_simplification": "engagement-regimen-v1",
}
UPLIFT_MODEL_NAMES = {
    "text_reminder":        "uplift-text-v2",
    "education":            "uplift-education-v1",
    "pharmacist_consult":   "uplift-pharmacist-v2",
    "cost_assistance":      "uplift-cost-assist-v2",
    "med_sync":             "uplift-med-sync-v1",
    "regimen_simplification": "uplift-regimen-v0",   # v0: still calibrating
}
SUPERVISED_BARRIER_MODEL_NAME = "barrier-supervised-v3"

# --- DynamoDB Table Names ---
# Six tables. Keep them separate so access patterns stay clean and
# IAM scoping is precise:
#   1. intervention-catalog:        canonical intervention record (intervention_id PK)
#   2. patient-profile:             member demographics, prefs, regimen meta (patient_id PK)
#   3. recommendation-log:          per (patient, intervention, medication, run_date) row
#   4. barrier-classifications:     per (patient, therapeutic_class) ranked-barriers row
#   5. engagement-events:           raw events (event_id PK; tracking_id GSI)
#   6. pcp-overrides:               PCP-declined recommendations for clinical review
INTERVENTION_CATALOG_TABLE     = "intervention-catalog"
PATIENT_PROFILE_TABLE          = "patient-profile"
RECOMMENDATION_LOG_TABLE       = "recommendation-log"
BARRIER_CLASSIFICATIONS_TABLE  = "barrier-classifications"
ENGAGEMENT_EVENTS_TABLE        = "engagement-events"
PCP_OVERRIDES_TABLE            = "pcp-overrides"

# --- S3 Buckets and Prefixes ---
# Production: each bucket has its own KMS key and bucket policy. The
# example uses placeholder names; replace with your account's buckets.
PHARMACY_DATA_LAKE_BUCKET      = "adherence-pharmacy-data-lake"
TARGET_SET_BUCKET              = "adherence-target-set"
CANDIDATES_BUCKET              = "adherence-candidates"
SCORES_BUCKET                  = "adherence-scores"
FEATURE_STORE_OFFLINE_BUCKET   = "adherence-feature-store-offline"
ATHENA_RESULTS_BUCKET          = "adherence-athena-results"

# --- Athena ---
ATHENA_WORKGROUP = "adherence-recommender"
ATHENA_DATABASE  = "adherence_data_lake"

# --- Kinesis ---
# Engagement stream pattern reused from Recipes 4.1-4.4, with new
# event types: pharmacy_fill_observed, refill_gap_detected,
# adherence_intervention_recommended, intervention_outreach_sent,
# intervention_engaged, intervention_completed, barrier_elicited,
# med_sync_enrolled, cost_assistance_initiated,
# cost_assistance_approved, pcp_override, pharmacist_consult_completed.
ENGAGEMENT_STREAM_NAME = "engagement-stream"

# --- SES ---
SES_FROM_ADDRESS         = "adherence@example-health-plan.org"
SES_CONFIGURATION_SET    = "adherence-baa"

# --- Tracked Therapeutic Classes ---
# The three CMS Star Ratings Part D adherence measures plus a few
# common chronic classes. Production: align to the current PQA
# (Pharmacy Quality Alliance) measure specifications, which CMS
# updates periodically.
TRACKED_THERAPEUTIC_CLASSES = [
    "statins",
    "ras_antagonists",      # ACE inhibitors, ARBs
    "oral_diabetes",        # metformin, SGLT2, DPP-4, sulfonylureas, etc.
    "anticoagulants",
    "respiratory_inhalers",
]

# --- PDC Computation ---
# 30-day settled lag: retail claims arrive on a 1-2 day lag, mail-order
# on a 5-14 day lag, and specialty up to 30. Computing PDC against
# claims that haven't all arrived yet produces noisy, biased numbers
# that systematically under-estimate adherence for mail-order users.
SETTLED_LAG_DAYS    = 30
PDC_WINDOW_365_DAYS = 365
PDC_WINDOW_90_DAYS  = 90

# --- Adherence Threshold ---
# CMS Star Ratings adherent-threshold: PDC >= 0.80. Below this, the
# patient is considered non-adherent for measurement purposes.
PDC_ADHERENT_THRESHOLD = 0.80

# --- Barrier Classifier Thresholds ---
# Tuned with a clinical pharmacist; production calibration uses the
# pharmacist-elicited-barrier dataset to tune precision/recall by
# barrier per therapeutic class.
COPAY_HIGH_THRESHOLD            = 35.0   # USD; above this, cost barrier is plausible
NEED_REVIEW_THRESHOLD           = 0.70   # need score above which LLM disagreement triggers review
LLM_LOW_CONFIDENCE_THRESHOLD    = 0.55   # blended top-1 below which LLM is consulted
SYMPTOMATIC_LATENT_CLASSES      = {"statins", "ras_antagonists"}

# Barrier taxonomy. Six categories; never collapse to a single label.
BARRIER_CATEGORIES = ["cost", "forgetfulness", "beliefs", "side_effects",
                      "complexity", "access"]

# --- Allocator Policy Weights ---
# Documented, version-controlled, reviewable. The cost_efficiency
# weight is what stops a $80 pharmacist consult from being chosen over
# a $0.05 reminder for every candidate where uplift is similar. The
# weights are policy: a cross-functional review (medical director,
# pharmacist lead, equity lead, data science) sets these.
POLICY_WEIGHTS = {
    "need":             0.25,
    "barrier_fit":      0.20,
    "engagement":       0.10,
    "uplift":           0.35,
    "cost_efficiency":  0.10,
}

# --- Per-Patient Caps ---
# Documented caps on how aggressively the recommender targets a single
# patient in one run. Without these, the optimization can stack 5
# interventions on the most-targetable patient and zero on patients
# with weaker model signal but real clinical need.
MAX_INTERVENTIONS_PER_PATIENT_PER_RUN  = 2
MAX_HIGH_TOUCH_PER_PATIENT_PER_RUN     = 1
MAX_CONTACTS_PER_PATIENT_30D           = 3

# --- Cost-efficiency Math Guard ---
# Avoid divide-by-zero when intervention cost is zero (e.g., a
# fully manufacturer-funded reminder).
MIN_COST_FOR_DIVISION = 0.01

# --- Run Configuration ---
POLICY_VERSION = "adherence-policy-v0.3"

# CloudWatch namespace for adherence metrics. Slice by intervention
# type, therapeutic class, language, engagement-history quartile, and
# SDOH cohort to catch subgroup drift.
METRIC_NAMESPACE = "AdherenceRecommender"

# Synthetic intervention catalog. In production this lives in DynamoDB
# and is updated by the catalog-sync Lambda when vendors push catalog
# changes through EventBridge. Each `supported_barriers` map gives the
# intervention's fit per barrier in [0, 1]; the barrier-fit score is
# the dot product of the patient's ranked-barriers vector with this
# map.
SAMPLE_INTERVENTIONS = [
    {
        "intervention_id":          "text-reminder-001",
        "type":                     "text_reminder",
        "display_name":             "Standard SMS Refill Reminder",
        "supported_barriers": {
            "forgetfulness":         1.0,
            "complexity":            0.4,
            "cost":                  0.0,
            "beliefs":               0.0,
            "side_effects":          0.0,
            "access":                0.0,
        },
        "marginal_cost":            0.05,        # cents per send
        "daily_capacity":           50000,        # essentially unbounded
        "is_high_touch":            False,
        "generates_patient_contact": True,
        "brand_only":               False,
        "requires_lis":             False,
        "requires_partner_pharmacy": False,
        "supported_languages":      ["en", "es"],
        "cooldown_days":            7,
        "default_template":         "reminder-default-en",
    },
    {
        "intervention_id":          "education-001",
        "type":                     "education",
        "display_name":             "Statin Education Module",
        "supported_barriers": {
            "beliefs":               0.8,
            "side_effects":          0.4,
            "forgetfulness":         0.1,
            "cost":                  0.0,
            "complexity":            0.0,
            "access":                0.0,
        },
        "marginal_cost":            0.50,
        "daily_capacity":           5000,
        "is_high_touch":            False,
        "generates_patient_contact": True,
        "brand_only":               False,
        "requires_lis":             False,
        "requires_partner_pharmacy": False,
        "supported_languages":      ["en", "es"],
        "cooldown_days":            30,
        "default_template":         "education-statin-default-en",
        "applicable_classes":       ["statins"],
    },
    {
        "intervention_id":          "pharmacist-consult-001",
        "type":                     "pharmacist_consult",
        "display_name":             "Telephonic Clinical Pharmacist Consult",
        "supported_barriers": {
            "beliefs":               0.9,
            "side_effects":          0.95,
            "complexity":            0.7,
            "cost":                  0.5,        # pharmacist can route to cost-assist
            "forgetfulness":         0.5,
            "access":                0.6,
        },
        "marginal_cost":            55.00,       # FTE time per substantive consult
        "daily_capacity":           80,           # 4 pharmacists * 20 consults/day
        "is_high_touch":            True,
        "generates_patient_contact": True,
        "brand_only":               False,
        "requires_lis":             False,
        "requires_partner_pharmacy": False,
        "supported_languages":      ["en", "es"],
        "cooldown_days":            90,
        "default_template":         "pharmacist-default",
    },
    {
        "intervention_id":          "cost-assist-001",
        "type":                     "cost_assistance",
        "display_name":             "Cost-Assistance Navigation",
        "supported_barriers": {
            "cost":                  1.0,
            "access":                0.3,
            "beliefs":               0.0,
            "side_effects":          0.0,
            "forgetfulness":         0.0,
            "complexity":            0.0,
        },
        "marginal_cost":            45.00,       # case-mgmt time per application
        "daily_capacity":           40,
        "is_high_touch":            True,
        "generates_patient_contact": True,
        "brand_only":               False,        # also covers formulary alternatives
        "requires_lis":             False,
        "requires_partner_pharmacy": False,
        "supported_languages":      ["en", "es"],
        "cooldown_days":            180,
        "default_template":         "cost-assist-default",
    },
    {
        "intervention_id":          "med-sync-001",
        "type":                     "med_sync",
        "display_name":             "Pharmacy Med-Sync Enrollment",
        "supported_barriers": {
            "complexity":            0.95,
            "forgetfulness":         0.6,
            "cost":                  0.1,
            "beliefs":               0.0,
            "side_effects":          0.0,
            "access":                0.4,
        },
        "marginal_cost":            8.00,         # one-time enrollment + ongoing
        "daily_capacity":           120,
        "is_high_touch":            False,
        "generates_patient_contact": True,
        "brand_only":               False,
        "requires_lis":             False,
        "requires_partner_pharmacy": True,
        "supported_languages":      ["en", "es"],
        "cooldown_days":            365,
        "default_template":         "med-sync-default",
    },
    {
        "intervention_id":          "regimen-simplify-001",
        "type":                     "regimen_simplification",
        "display_name":             "PCP Regimen Simplification Referral",
        "supported_barriers": {
            "complexity":            0.95,
            "forgetfulness":         0.5,
            "side_effects":          0.3,
            "beliefs":               0.1,
            "cost":                  0.0,
            "access":                0.0,
        },
        "marginal_cost":            12.00,        # care-team coordination time
        "daily_capacity":           50,
        "is_high_touch":            False,
        "generates_patient_contact": False,       # PCP-mediated, not patient-direct
        "brand_only":               False,
        "requires_lis":             False,
        "requires_partner_pharmacy": False,
        "supported_languages":      ["en", "es"],
        "cooldown_days":            180,
        "default_template":         "regimen-simplify-default",
    },
]

# Partner-pharmacy IDs that support med-sync via API. Production: a
# DynamoDB lookup or a small config table maintained by the contracts
# team. Pharmacy chains and PSAOs join and leave the partnership list
# at quarterly cadence; treat the list as data, not code.
PARTNER_PHARMACIES = {"PHARM-CHAIN-A", "PHARM-CHAIN-B", "PHARM-PSAO-1"}

# Cohort axes for equity-floor matching. Same shape as 4.4.
EQUITY_FLOORS = {
    "pharmacist-consult-001": {
        "engagement_q1":         8,    # lowest engagement quartile
        "language_non_en":       6,
        "sdoh_low_food_security": 6,
    },
    "cost-assist-001": {
        "sdoh_low_food_security": 8,
        "language_non_en":       4,
    },
    "education-001": {
        "language_non_en":       50,
    },
    # Reminder, med-sync, regimen-simplification: no floor in this
    # example. Production may add equity floors based on observed
    # disparities in the engagement and outcome dashboards.
}

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

def _today_str() -> str:
    """Current UTC date as YYYY-MM-DD string for run_date."""
    return datetime.datetime.now(timezone.utc).date().isoformat()

def _emit_metric(name: str, value: float, dimensions: dict) -> None:
    """
    Emit a CloudWatch custom metric. Swallows errors so a metric-publish
    failure never breaks the recommendation pipeline. CloudWatch metric
    publishing is best-effort observability, not a correctness boundary.
    """
    try:
        cloudwatch_client.put_metric_data(
            Namespace=METRIC_NAMESPACE,
            MetricData=[{
                "MetricName": name,
                "Dimensions": [
                    {"Name": k, "Value": str(v)[:255]} for k, v in dimensions.items()
                ],
                "Value": float(value),
                "Unit":  "Count",
            }],
        )
    except Exception as exc:
        logger.warning("Metric publish failed for %s: %s", name, exc)

def _to_decimal(value) -> Decimal:
    """
    DynamoDB does not accept Python floats. Going through str avoids
    binary-precision issues. Wrap floats at the persistence boundary
    and forget about it. (This is the SDK gotcha that bites every
    boto3 newcomer; fixed at the boundary, not in business logic.)
    """
    if isinstance(value, Decimal):
        return value
    return Decimal(str(value))

def _to_decimal_dict(d: dict) -> dict:
    """Recursively convert numeric values in a dict to Decimal for DynamoDB."""
    out = {}
    for k, v in d.items():
        if isinstance(v, (int, float)) and not isinstance(v, bool):
            out[k] = _to_decimal(v)
        elif isinstance(v, dict):
            out[k] = _to_decimal_dict(v)
        elif isinstance(v, list):
            out[k] = [_to_decimal_dict(x) if isinstance(x, dict)
                      else _to_decimal(x) if isinstance(x, (int, float)) and not isinstance(x, bool)
                      else x
                      for x in v]
        else:
            out[k] = v
    return out

def _from_decimal(value):
    """Inverse of _to_decimal for reading DynamoDB items into Python."""
    if isinstance(value, Decimal):
        return float(value)
    if isinstance(value, dict):
        return {k: _from_decimal(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_from_decimal(v) for v in value]
    return value

def _make_tracking_id(run_date: str, patient_id: str,
                      therapeutic_class: str, intervention_id: str) -> str:
    """
    Tracking_id used to join recommendation -> outreach -> engagement.

    NOTE: This example uses a readable string for clarity. Production
    must replace this with an opaque, non-reversible identifier (UUID
    or HMAC over the composite). Plain-text patient_id and
    therapeutic_class embedded in tracking IDs (carried in email
    open-tracking pixels, SMS click-through links, vendor outreach
    platform handoffs) are PHI leakage. The "Gap to Production"
    section at the end of this file flags the same issue.
    """
    return f"adherence-{run_date}-{patient_id}-{therapeutic_class}-{intervention_id}"

def _parse_s3_uri(uri: str) -> tuple:
    if not uri.startswith("s3://"):
        raise ValueError(f"Not an S3 URI: {uri}")
    parts = uri[5:].split("/", 1)
    return parts[0], parts[1] if len(parts) > 1 else ""

def _wait_for_athena_query(execution_id: str, timeout_seconds: int = 300) -> None:
    """Poll Athena until the query reaches a terminal state."""
    start = time.time()
    while True:
        response = athena_client.get_query_execution(QueryExecutionId=execution_id)
        state = response["QueryExecution"]["Status"]["State"]
        if state == "SUCCEEDED":
            return
        if state in ("FAILED", "CANCELLED"):
            reason = response["QueryExecution"]["Status"].get("StateChangeReason", "")
            raise RuntimeError(f"Athena query {execution_id} {state}: {reason}")
        if time.time() - start > timeout_seconds:
            raise TimeoutError(f"Athena query {execution_id} timed out after {timeout_seconds}s")
        time.sleep(2)

def _wait_for_transform_job(job_name: str, timeout_seconds: int = 3600) -> None:
    """Poll a Batch Transform job until it reaches a terminal state."""
    start = time.time()
    while True:
        response = sagemaker_client.describe_transform_job(TransformJobName=job_name)
        status = response["TransformJobStatus"]
        if status == "Completed":
            return
        if status in ("Failed", "Stopped"):
            reason = response.get("FailureReason", "")
            raise RuntimeError(f"Transform job {job_name} {status}: {reason}")
        if time.time() - start > timeout_seconds:
            raise TimeoutError(f"Transform job {job_name} timed out")
        time.sleep(15)

def compute_adherence_features(run_date: str) -> str:
    """
    Run the adherence-computation pipeline:

      1. Pull settled fill claims (Athena) for the trailing 365 days
         ending at run_date - SETTLED_LAG_DAYS.
      2. For each (patient_id, therapeutic_class), build the covered-
         days set, compute PDC at 365d and 90d, compute fill cadence
         and copay-paid statistics, assess data quality.
      3. Write the per-medication features to SageMaker Feature Store
         (online + offline) for downstream consumption.

    Returns the offline Parquet path that downstream stages can use
    when Feature Store online lookups aren't appropriate (large
    batch joins).
    """
    settled_window_end = (
        datetime.date.fromisoformat(run_date) - datetime.timedelta(days=SETTLED_LAG_DAYS)
    )
    settled_window_start = settled_window_end - datetime.timedelta(days=PDC_WINDOW_365_DAYS)

    classes_quoted = ", ".join(f"'{c}'" for c in TRACKED_THERAPEUTIC_CLASSES)
    query = f"""
        SELECT
            patient_id,
            ndc,
            therapeutic_class,
            fill_date,
            days_supply,
            channel,
            copay_paid,
            was_cash_pay,
            pharmacy_id
        FROM pharmacy_claims
        WHERE fill_date BETWEEN DATE '{settled_window_start.isoformat()}'
                            AND DATE '{settled_window_end.isoformat()}'
          AND therapeutic_class IN ({classes_quoted})
    """.strip()

    logger.info(
        "Pulling settled fills for run_date=%s window=%s..%s",
        run_date, settled_window_start, settled_window_end,
    )
    execution = athena_client.start_query_execution(
        QueryString=query,
        QueryExecutionContext={"Database": ATHENA_DATABASE},
        WorkGroup=ATHENA_WORKGROUP,
        ResultConfiguration={
            "OutputLocation": f"s3://{ATHENA_RESULTS_BUCKET}/fills/{run_date}/",
        },
    )
    _wait_for_athena_query(execution["QueryExecutionId"], timeout_seconds=600)

    # Production: a Glue job streams the Athena results into an
    # in-memory frame keyed by (patient_id, therapeutic_class). The
    # example uses an injectable loader so tests can supply synthetic
    # fills without round-tripping through Athena.
    fills_by_patient_class = _load_settled_fills(execution["QueryExecutionId"], run_date)

    feature_records = []
    for (patient_id, therapeutic_class), fills in fills_by_patient_class.items():
        adherence = _compute_pdc_for_class(fills, settled_window_end)
        feature_records.append({
            "patient_id":         patient_id,
            "therapeutic_class":  therapeutic_class,
            "run_date":           run_date,
            **adherence,
        })

    _persist_adherence_features_to_feature_store(feature_records, run_date)

    # Aggregate per-patient regimen features (across all chronic
    # medications). The barrier classifier and complexity-related
    # interventions consume these.
    regimen_records = _compute_regimen_features(fills_by_patient_class, run_date)
    _persist_regimen_features_to_feature_store(regimen_records, run_date)

    offline_path = (
        f"s3://{FEATURE_STORE_OFFLINE_BUCKET}/run_date={run_date}/"
        f"adherence-features.parquet"
    )
    logger.info(
        "Computed adherence features for %d (patient, class) pairs; offline=%s",
        len(feature_records), offline_path,
    )
    return offline_path

def _compute_pdc_for_class(fills: list, settled_window_end: datetime.date) -> dict:
    """
    Build the carry-forward covered-days set for one (patient,
    therapeutic_class) and compute PDC at 365d and 90d.

    Each fill contributes a half-open interval
    [fill_date, fill_date + days_supply). Overlapping fills (early
    refills, stockpiling) extend the covered set without
    double-counting. The set membership test for "is day d covered"
    is what makes mail-order, retail, and synchronized refills all
    produce the same PDC for the same actual adherence.
    """
    if not fills:
        return _empty_pdc_record()

    # Sort by fill_date so the carry-forward loop is monotone.
    fills_sorted = sorted(fills, key=lambda f: f["fill_date"])

    covered_days = set()
    for fill in fills_sorted:
        fd = fill["fill_date"]
        if isinstance(fd, str):
            fd = datetime.date.fromisoformat(fd)
        for offset in range(int(fill["days_supply"])):
            covered_days.add(fd + datetime.timedelta(days=offset))

    # Restrict to days at or before the settled window end. A fill
    # with days_supply that runs past the window doesn't credit days
    # outside the window.
    window_start_365 = settled_window_end - datetime.timedelta(days=PDC_WINDOW_365_DAYS - 1)
    window_start_90  = settled_window_end - datetime.timedelta(days=PDC_WINDOW_90_DAYS - 1)

    in_window_365 = sum(
        1 for d in covered_days
        if window_start_365 <= d <= settled_window_end
    )
    in_window_90 = sum(
        1 for d in covered_days
        if window_start_90 <= d <= settled_window_end
    )

    pdc_365 = in_window_365 / PDC_WINDOW_365_DAYS
    pdc_90  = in_window_90  / PDC_WINDOW_90_DAYS

    last_fill = fills_sorted[-1]
    last_fill_date = last_fill["fill_date"]
    if isinstance(last_fill_date, str):
        last_fill_date = datetime.date.fromisoformat(last_fill_date)
    last_days_supply = int(last_fill["days_supply"])

    days_since_last_fill = (settled_window_end - last_fill_date).days
    gap_days = max(0, days_since_last_fill - last_days_supply)

    # Channel mix and copay-paid statistics. Proxies the barrier
    # classifier consumes: a patient on a mostly-mail-order channel
    # has a different barrier signature than one fragmented across
    # retail pharmacies. A patient with rising copays may be
    # cost-barriered; a patient with stable copays whose adherence
    # dropped probably isn't.
    channel_counts = {}
    copay_values = []
    cash_pay_count = 0
    for fill in fills_sorted:
        channel_counts[fill["channel"]] = channel_counts.get(fill["channel"], 0) + 1
        if fill.get("copay_paid") is not None:
            copay_values.append(float(fill["copay_paid"]))
        if fill.get("was_cash_pay"):
            cash_pay_count += 1
    total_fills = len(fills_sorted)
    channel_mix = {ch: count / total_fills for ch, count in channel_counts.items()}

    copay_paid_p50 = _percentile(copay_values, 50) if copay_values else 0.0
    copay_paid_p90 = _percentile(copay_values, 90) if copay_values else 0.0

    # Detect the most-recent copay change: did the most recent fill's
    # copay differ from the prior fills' typical copay? Used by the
    # cost-barrier rule.
    most_recent_copay = float(last_fill.get("copay_paid") or 0.0)
    prior_copays = [float(f.get("copay_paid") or 0.0) for f in fills_sorted[:-1]]
    gap_onset_aligned_with_copay_change = False
    if prior_copays:
        prior_p50 = _percentile(prior_copays, 50)
        if most_recent_copay > prior_p50 * 1.5 and most_recent_copay > COPAY_HIGH_THRESHOLD:
            gap_onset_aligned_with_copay_change = True

    pharmacy_count = len({f["pharmacy_id"] for f in fills_sorted})
    data_quality_flag = _assess_data_quality(
        fills_sorted, channel_mix, pharmacy_count, cash_pay_count, total_fills,
    )

    fills_pattern_is_sporadic = _is_sporadic_pattern(fills_sorted)
    fills_pattern_is_consistent_then_stopped = _is_consistent_then_stopped(
        fills_sorted, settled_window_end,
    )
    discontinued_after_one_or_two_fills = total_fills <= 2 and gap_days > 60

    return {
        "pdc_365":                                  pdc_365,
        "pdc_90":                                   pdc_90,
        "gap_days":                                 gap_days,
        "last_fill_date":                           last_fill_date.isoformat(),
        "channel_mix":                              channel_mix,
        "copay_paid_p50":                           copay_paid_p50,
        "copay_paid_p90":                           copay_paid_p90,
        "most_recent_copay":                        most_recent_copay,
        "gap_onset_aligned_with_copay_change":      gap_onset_aligned_with_copay_change,
        "cash_pay_indicator":                       cash_pay_count > 0,
        "pharmacy_count":                           pharmacy_count,
        "fills_pattern_is_sporadic":                fills_pattern_is_sporadic,
        "fills_pattern_is_consistent_then_stopped": fills_pattern_is_consistent_then_stopped,
        "discontinued_after_one_or_two_fills":      discontinued_after_one_or_two_fills,
        "total_fills":                              total_fills,
        "data_quality_flag":                        data_quality_flag,
    }

def _empty_pdc_record() -> dict:
    """Sentinel record for a (patient, class) with no fills."""
    return {
        "pdc_365": 0.0, "pdc_90": 0.0, "gap_days": 999,
        "last_fill_date": None, "channel_mix": {},
        "copay_paid_p50": 0.0, "copay_paid_p90": 0.0,
        "most_recent_copay": 0.0,
        "gap_onset_aligned_with_copay_change": False,
        "cash_pay_indicator": False, "pharmacy_count": 0,
        "fills_pattern_is_sporadic": False,
        "fills_pattern_is_consistent_then_stopped": False,
        "discontinued_after_one_or_two_fills": False,
        "total_fills": 0,
        "data_quality_flag": "no_history",
    }

def _percentile(values: list, pct: int) -> float:
    """Simple percentile (numpy not required for the demo)."""
    if not values:
        return 0.0
    sorted_v = sorted(values)
    idx = int(len(sorted_v) * pct / 100)
    idx = min(idx, len(sorted_v) - 1)
    return sorted_v[idx]

def _is_sporadic_pattern(fills_sorted: list) -> bool:
    """
    Sporadic = fill-to-fill gaps vary widely (some on time, some
    months late). Distinguishes forgetful patients from
    consistently-non-adherent patients.
    """
    if len(fills_sorted) < 3:
        return False
    gaps = []
    for i in range(1, len(fills_sorted)):
        d_curr = fills_sorted[i]["fill_date"]
        d_prev = fills_sorted[i - 1]["fill_date"]
        if isinstance(d_curr, str):
            d_curr = datetime.date.fromisoformat(d_curr)
        if isinstance(d_prev, str):
            d_prev = datetime.date.fromisoformat(d_prev)
        gaps.append((d_curr - d_prev).days)
    if not gaps:
        return False
    mean_gap = sum(gaps) / len(gaps)
    if mean_gap == 0:
        return False
    variance = sum((g - mean_gap) ** 2 for g in gaps) / len(gaps)
    cv = (variance ** 0.5) / mean_gap
    return cv > 0.5

def _is_consistent_then_stopped(fills_sorted: list,
                                 settled_window_end: datetime.date) -> bool:
    """
    Consistent-then-stopped = patient filled regularly for a stretch,
    then stopped entirely. Suggests beliefs/concerns or access
    barriers, not forgetfulness.
    """
    if len(fills_sorted) < 3:
        return False
    last_fd = fills_sorted[-1]["fill_date"]
    if isinstance(last_fd, str):
        last_fd = datetime.date.fromisoformat(last_fd)
    days_since = (settled_window_end - last_fd).days
    if days_since < 90:
        return False
    # Check the prior fills had small gaps.
    early_gaps = []
    for i in range(1, min(len(fills_sorted), 5)):
        d_curr = fills_sorted[i]["fill_date"]
        d_prev = fills_sorted[i - 1]["fill_date"]
        if isinstance(d_curr, str):
            d_curr = datetime.date.fromisoformat(d_curr)
        if isinstance(d_prev, str):
            d_prev = datetime.date.fromisoformat(d_prev)
        early_gaps.append((d_curr - d_prev).days)
    if not early_gaps:
        return False
    return max(early_gaps) < 45

def _assess_data_quality(fills_sorted: list, channel_mix: dict,
                          pharmacy_count: int, cash_pay_count: int,
                          total_fills: int) -> str:
    """
    Tag each (patient, class) with a data-quality flag downstream
    consumers can gate on. A confident "non-adherent" label on a
    patient with `cash_pay_partial` is a confidently wrong label.
    """
    if total_fills == 0:
        return "no_history"
    if cash_pay_count > 0 and cash_pay_count / total_fills > 0.3:
        return "cash_pay_partial"
    if pharmacy_count >= 3:
        return "multi_pharmacy_fragmented"
    if total_fills < 3:
        return "sparse_history"
    return "complete"

def _load_settled_fills(athena_execution_id: str, run_date: str) -> dict:
    """
    Load Athena results into an in-memory dict keyed on
    (patient_id, therapeutic_class). Production: a Glue job that
    writes parquet partitioned by therapeutic_class. The example
    returns an empty dict; the demo runner monkey-patches this with
    synthetic fills for the offline run.
    """
    return {}

def _persist_adherence_features_to_feature_store(records: list, run_date: str) -> None:
    """
    Push per-(patient, class) feature records to the
    `patient-medication-adherence` SageMaker Feature Store group.

    Production wires this through the Feature Store's PutRecord API
    (online store) and lets the offline-store ingestion handle the
    S3-Parquet write asynchronously. The example logs the call shape
    so the demo can run without a real feature group provisioned.
    """
    for r in records:
        try:
            sagemaker_featurestore.put_record(
                FeatureGroupName="patient-medication-adherence",
                Record=[
                    {"FeatureName": "patient_id",
                     "ValueAsString": r["patient_id"]},
                    {"FeatureName": "therapeutic_class",
                     "ValueAsString": r["therapeutic_class"]},
                    {"FeatureName": "pdc_365",
                     "ValueAsString": str(r["pdc_365"])},
                    {"FeatureName": "pdc_90",
                     "ValueAsString": str(r["pdc_90"])},
                    {"FeatureName": "gap_days",
                     "ValueAsString": str(r["gap_days"])},
                    {"FeatureName": "data_quality_flag",
                     "ValueAsString": r["data_quality_flag"]},
                    {"FeatureName": "event_time",
                     "ValueAsString": _now_iso()},
                ],
            )
        except Exception as exc:
            # In the offline demo the feature group doesn't exist;
            # log and continue. Production: alarm on persistent
            # PutRecord failures.
            logger.debug(
                "Feature Store PutRecord skipped for %s/%s: %s",
                r["patient_id"], r["therapeutic_class"], exc,
            )

def _compute_regimen_features(fills_by_patient_class: dict, run_date: str) -> list:
    """
    Aggregate per-patient regimen features across all chronic
    medications. The barrier classifier consumes regimen size and
    pharmacy count to detect complexity-driven non-adherence.
    """
    by_patient = {}
    for (patient_id, therapeutic_class), fills in fills_by_patient_class.items():
        entry = by_patient.setdefault(patient_id, {
            "patient_id": patient_id,
            "classes":    set(),
            "pharmacies": set(),
            "fills":      0,
        })
        entry["classes"].add(therapeutic_class)
        entry["fills"] += len(fills)
        for f in fills:
            entry["pharmacies"].add(f["pharmacy_id"])

    profile_table = dynamodb.Table(PATIENT_PROFILE_TABLE)
    records = []
    for patient_id, entry in by_patient.items():
        try:
            response = profile_table.get_item(Key={"patient_id": patient_id})
            profile = _from_decimal(response.get("Item") or {})
        except Exception:
            profile = {}
        records.append({
            "patient_id":           patient_id,
            "regimen_size":         len(entry["classes"]),
            "num_classes_tracked":  len(entry["classes"]),
            "pharmacy_count":       len(entry["pharmacies"]),
            "med_sync_enrolled":    bool(profile.get("med_sync_enrolled", False)),
            "doses_per_day_total":  profile.get("doses_per_day_total", 0),
            "run_date":             run_date,
        })
    return records

def _persist_regimen_features_to_feature_store(records: list, run_date: str) -> None:
    """Counterpart to the per-medication persistence; same caveats."""
    for r in records:
        try:
            sagemaker_featurestore.put_record(
                FeatureGroupName="patient-regimen",
                Record=[
                    {"FeatureName": "patient_id",         "ValueAsString": r["patient_id"]},
                    {"FeatureName": "regimen_size",       "ValueAsString": str(r["regimen_size"])},
                    {"FeatureName": "pharmacy_count",     "ValueAsString": str(r["pharmacy_count"])},
                    {"FeatureName": "med_sync_enrolled",  "ValueAsString": str(r["med_sync_enrolled"])},
                    {"FeatureName": "event_time",         "ValueAsString": _now_iso()},
                ],
            )
        except Exception as exc:
            logger.debug(
                "Feature Store PutRecord skipped for regimen/%s: %s",
                r["patient_id"], exc,
            )

def classify_barriers_for_target_set(
    target_set: list,
    adherence_features: dict,
    regimen_features: dict,
    member_profiles: dict,
    run_date: str,
) -> dict:
    """
    Classify barriers for each (patient_id, therapeutic_class) in
    target_set.

    target_set:           list of dicts with patient_id, therapeutic_class
    adherence_features:   dict[(patient_id, therapeutic_class)] -> adherence_record
    regimen_features:     dict[patient_id] -> regimen_record
    member_profiles:      dict[patient_id] -> profile dict

    Returns dict[(patient_id, therapeutic_class)] -> ranked_barriers list.
    """
    barriers_table = dynamodb.Table(BARRIER_CLASSIFICATIONS_TABLE)
    results = {}

    for entry in target_set:
        patient_id = entry["patient_id"]
        therapeutic_class = entry["therapeutic_class"]
        adherence = adherence_features.get((patient_id, therapeutic_class))
        regimen = regimen_features.get(patient_id, {})
        member = member_profiles.get(patient_id, {})

        if adherence is None:
            logger.warning(
                "No adherence features for %s/%s; skipping barrier classification",
                patient_id, therapeutic_class,
            )
            continue

        # Stage A: rule-based pass.
        rule_results = _rule_based_barrier_classifier(
            adherence, regimen, member, therapeutic_class,
        )

        # Stage B: supervised refinement (where labels exist).
        # Production: invoke a SageMaker endpoint with the feature
        # vector; the classifier returns probabilities over the six
        # barrier classes. The example skips the live call.
        supervised_probs = _supervised_barrier_predict(
            adherence, regimen, member, therapeutic_class,
        )

        # Blend rule-based and supervised. Rule-based gets the heavier
        # weight because it's auditable; supervised is augmentation.
        blended = _blend_rule_and_supervised(
            rule_results, supervised_probs,
            rule_weight=0.6, supervised_weight=0.4,
        )

        # Compute the need score (used to decide whether to escalate
        # to LLM second opinion). Production: a real per-class need
        # model; the example uses a derivable proxy.
        need_score = _proxy_need_score(adherence, member, therapeutic_class)

        # Stage C: LLM second opinion for high-stakes ambiguous cases.
        llm_review = None
        if _should_invoke_llm(blended, need_score):
            try:
                llm_review = _bedrock_barrier_second_opinion(
                    adherence, regimen, member, therapeutic_class,
                )
                # Flag to pharmacist review if LLM disagrees with
                # blended top-1 on a high-need case.
                if (llm_review and llm_review.get("predicted_barrier")
                    and llm_review["predicted_barrier"] != blended["ranked_list"][0]["barrier"]
                    and need_score > NEED_REVIEW_THRESHOLD):
                    _flag_for_pharmacist_review(
                        patient_id, therapeutic_class, blended, llm_review,
                    )
            except Exception as exc:
                logger.warning(
                    "Bedrock barrier second opinion failed for %s/%s: %s",
                    patient_id, therapeutic_class, exc,
                )

        ranked = blended["ranked_list"]
        results[(patient_id, therapeutic_class)] = ranked

        # Persist the ranked barriers. Don't collapse to a single label.
        try:
            barriers_table.put_item(Item=_to_decimal_dict({
                "patient_id":         patient_id,
                "therapeutic_class":  therapeutic_class,
                "run_date":           run_date,
                "ranked_barriers":    ranked,
                "llm_second_opinion": llm_review,
                "data_quality_flag":  adherence["data_quality_flag"],
                "classifier_versions": {
                    "rules":      "rules-v3",
                    "supervised": SUPERVISED_BARRIER_MODEL_NAME,
                    "llm":        BARRIER_CLASSIFIER_MODEL_ID,
                },
                "created_at":         _now_iso(),
            }))
        except Exception as exc:
            logger.warning(
                "Failed to persist barriers for %s/%s: %s",
                patient_id, therapeutic_class, exc,
            )

    logger.info(
        "Classified barriers for %d (patient, class) pairs", len(results),
    )
    return results

def _rule_based_barrier_classifier(
    adherence: dict, regimen: dict, member: dict, therapeutic_class: str,
) -> list:
    """
    Deterministic rule-based barrier classifier. Each rule emits one
    or more {barrier, rule_confidence} dicts. Multiple rules can fire.

    Production: rules live in a versioned config (DynamoDB or a
    Git-tracked YAML), reviewed by a clinical pharmacist. Hard-coding
    them here for clarity.
    """
    rule_results = []

    # Cost barrier: high copay aligned with gap onset.
    if (adherence["gap_days"] > 30
        and adherence["most_recent_copay"] > COPAY_HIGH_THRESHOLD
        and adherence["gap_onset_aligned_with_copay_change"]):
        rule_results.append({"barrier": "cost", "rule_confidence": 0.85})

    # Cost barrier: high copay without LIS enrollment, even without
    # alignment to a copay change.
    if (adherence["most_recent_copay"] > COPAY_HIGH_THRESHOLD
        and not adherence["gap_onset_aligned_with_copay_change"]
        and not member.get("lis_enrolled", False)):
        rule_results.append({"barrier": "cost", "rule_confidence": 0.55})

    # Complexity / forgetfulness from regimen size + sporadic fills.
    if (regimen.get("regimen_size", 0) >= 4
        and adherence["fills_pattern_is_sporadic"]
        and not regimen.get("med_sync_enrolled", False)):
        rule_results.append({"barrier": "complexity",    "rule_confidence": 0.65})
        rule_results.append({"barrier": "forgetfulness", "rule_confidence": 0.45})

    # Side effects: gap started after a side-effect-coded encounter.
    if member.get("recent_side_effect_encounter") and member.get(
        "side_effect_encounter_before_gap_onset", False
    ):
        rule_results.append({"barrier": "side_effects", "rule_confidence": 0.80})

    # Beliefs: classic asymptomatic-condition discontinuation pattern.
    if (therapeutic_class in SYMPTOMATIC_LATENT_CLASSES
        and adherence["discontinued_after_one_or_two_fills"]):
        rule_results.append({"barrier": "beliefs", "rule_confidence": 0.55})

    # Access: consistent then stopped, no recent care interaction.
    if (adherence["fills_pattern_is_consistent_then_stopped"]
        and not member.get("recent_side_effect_encounter")
        and member.get("recent_pcp_encounter_count", 0) == 0):
        rule_results.append({"barrier": "access", "rule_confidence": 0.40})

    # Default: if no rule fired, treat as low-confidence forgetfulness.
    # Better than empty so downstream barrier-fit scoring has something
    # to consume; the supervised classifier will refine.
    if not rule_results:
        rule_results.append({"barrier": "forgetfulness", "rule_confidence": 0.30})

    return rule_results

def _supervised_barrier_predict(
    adherence: dict, regimen: dict, member: dict, therapeutic_class: str,
) -> dict:
    """
    Invoke the supervised barrier classifier (SageMaker endpoint or
    Batch Transform). Returns a probability distribution over the six
    barrier categories.

    Production: the model is trained on (features, elicited_barrier)
    labels from pharmacist consults. Without those labels, this stage
    is omitted and the recommender runs on rule-based output only.

    The example returns a uniform-ish distribution as a placeholder;
    the demo runner can monkey-patch with synthetic predictions.
    """
    # Placeholder distribution. Avoid uniform exactly so the blend
    # function still has something to combine.
    return {
        "cost":          0.18,
        "forgetfulness": 0.20,
        "beliefs":       0.17,
        "side_effects":  0.15,
        "complexity":    0.16,
        "access":        0.14,
    }

def _blend_rule_and_supervised(
    rule_results: list, supervised_probs: dict,
    rule_weight: float, supervised_weight: float,
) -> dict:
    """
    Combine rule-based and supervised outputs into a single ranked
    list of barriers with confidences and source attribution.

    For each barrier:
      - rule_part = max rule_confidence across rules that fired for it (or 0)
      - supervised_part = supervised probability for that barrier
      - blended_confidence = rule_weight * rule_part + supervised_weight * supervised_part
    """
    rule_max = {b: 0.0 for b in BARRIER_CATEGORIES}
    rule_sources = {b: [] for b in BARRIER_CATEGORIES}
    for entry in rule_results:
        b = entry["barrier"]
        if entry["rule_confidence"] > rule_max[b]:
            rule_max[b] = entry["rule_confidence"]
        rule_sources[b].append("rule")

    ranked = []
    for b in BARRIER_CATEGORIES:
        rule_part = rule_max[b]
        sup_part = supervised_probs.get(b, 0.0)
        confidence = rule_weight * rule_part + supervised_weight * sup_part
        sources = []
        if rule_part > 0:
            sources.append("rule")
        if sup_part > 0:
            sources.append("supervised")
        ranked.append({
            "barrier":    b,
            "confidence": round(confidence, 4),
            "sources":    sources,
        })

    ranked.sort(key=lambda x: x["confidence"], reverse=True)
    return {
        "ranked_list": ranked,
        "top_1":       ranked[0],
    }

def _proxy_need_score(adherence: dict, member: dict, therapeutic_class: str) -> float:
    """
    Cheap proxy for the per-class need score so we can decide whether
    to spend an LLM call on this case. Production replaces this with
    a SageMaker Batch Transform output read.

    Higher score = greater clinical risk of continued non-adherence.
    """
    score = 0.0
    pdc_365 = adherence.get("pdc_365", 1.0)
    if pdc_365 < 0.50:
        score += 0.4
    elif pdc_365 < 0.80:
        score += 0.25
    if therapeutic_class in {"statins", "ras_antagonists", "oral_diabetes"}:
        score += 0.2
    if member.get("age", 0) >= 65:
        score += 0.1
    if member.get("recent_hospitalization_for_class", False):
        score += 0.2
    return min(score, 1.0)

def _should_invoke_llm(blended: dict, need_score: float) -> bool:
    """
    Spend the LLM call only when:
      - blended top-1 confidence is low (< threshold), OR
      - need score is high enough that getting it right matters.
    """
    top_conf = blended["top_1"]["confidence"]
    if top_conf < LLM_LOW_CONFIDENCE_THRESHOLD:
        return True
    if need_score > NEED_REVIEW_THRESHOLD and top_conf < 0.80:
        return True
    return False

def _bedrock_barrier_second_opinion(
    adherence: dict, regimen: dict, member: dict, therapeutic_class: str,
) -> dict:
    """
    Invoke Bedrock to produce a structured-output barrier prediction.

    IMPORTANT: pass de-identified context to the LLM. Don't pass raw
    patient_id, name, phone, or NDC into the prompt; the LLM doesn't
    need them, and stripping them at this boundary limits any vendor-
    side logging exposure (Bedrock service terms commit to not using
    prompts to train foundation models, but defense-in-depth still
    applies).
    """
    # Build a structured, de-identified context block.
    context = {
        "therapeutic_class":      therapeutic_class,
        "pdc_365":                round(adherence["pdc_365"], 3),
        "pdc_90":                 round(adherence["pdc_90"], 3),
        "gap_days":               adherence["gap_days"],
        "most_recent_copay":      adherence["most_recent_copay"],
        "copay_p50":              adherence["copay_paid_p50"],
        "fills_pattern_is_sporadic":               adherence["fills_pattern_is_sporadic"],
        "fills_pattern_is_consistent_then_stopped": adherence["fills_pattern_is_consistent_then_stopped"],
        "discontinued_after_one_or_two_fills":     adherence["discontinued_after_one_or_two_fills"],
        "regimen_size":           regimen.get("regimen_size", 0),
        "med_sync_enrolled":      regimen.get("med_sync_enrolled", False),
        "lis_enrolled":           member.get("lis_enrolled", False),
        "age_band":               member.get("age_band"),
        "data_quality_flag":      adherence["data_quality_flag"],
    }

    prompt = f"""You are a clinical adherence specialist reviewing a patient's
medication adherence pattern. Identify the most likely barrier to adherence.

Allowed barrier categories: cost, forgetfulness, beliefs, side_effects,
complexity, access.

Observed signals:
{json.dumps(context, indent=2)}

Return ONLY valid JSON with this shape (no prose, no code fences):
{{
  "predicted_barrier":     "<one of the six categories>",
  "confidence":            "<low | moderate | high>",
  "rationale":             "<2-3 sentences citing specific observed signals>",
  "alternative_barriers":  ["<barrier>", "<barrier>"],
  "uncertainty_notes":     "<what would change your prediction>"
}}

Cite only signals from the Observed signals block above. Do not invent
clinical history that isn't shown.
"""
    response = bedrock_runtime.invoke_model(
        modelId=BARRIER_CLASSIFIER_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens":        500,
            "temperature":       0.0,    # deterministic for second opinion
            "messages":          [{"role": "user", "content": prompt}],
        }),
    )
    payload = json.loads(response["body"].read())
    completion = payload["content"][0]["text"]

    # Defensive JSON extraction.
    import re as _re
    match = _re.search(r"\{.*\}", completion, _re.DOTALL)
    if not match:
        raise ValueError("LLM returned no JSON object")
    parsed = json.loads(match.group(0))

    # Validate the structured output: barrier must be in allowed
    # taxonomy, confidence must be one of the allowed strings.
    if parsed.get("predicted_barrier") not in BARRIER_CATEGORIES:
        raise ValueError(
            f"LLM returned invalid barrier: {parsed.get('predicted_barrier')}"
        )
    if parsed.get("confidence") not in ("low", "moderate", "high"):
        raise ValueError(
            f"LLM returned invalid confidence: {parsed.get('confidence')}"
        )
    return parsed

def _flag_for_pharmacist_review(
    patient_id: str, therapeutic_class: str,
    blended: dict, llm_review: dict,
) -> None:
    """
    When the LLM disagrees materially with the blended classifier on
    a high-need case, flag for pharmacist review rather than silently
    picking one. Production: write to a `pharmacist-review-queue`
    table that the operations team consumes daily.
    """
    logger.info(
        "Pharmacist review flagged: patient=%s class=%s blended_top=%s llm_top=%s",
        patient_id, therapeutic_class,
        blended["top_1"]["barrier"], llm_review["predicted_barrier"],
    )

def build_candidate_triples(
    target_set: list,
    intervention_catalog: list,
    barriers_by_class: dict,
    member_profiles: dict,
    medication_metadata: dict,
    run_date: str,
) -> list:
    """
    Cross-product target_set with the intervention catalog, applying
    eligibility filters. Returns a flat list of candidate dicts with
    enough context that downstream scoring doesn't need to re-look-up
    member or medication state.

    barriers_by_class:    dict[(patient_id, therapeutic_class)] -> ranked_barriers
    medication_metadata:  dict[(patient_id, therapeutic_class)] -> {
                            ndc, is_brand, formulary_tier,
                            data_quality_flag, ...
                          }
    """
    candidates = []
    for entry in target_set:
        patient_id = entry["patient_id"]
        therapeutic_class = entry["therapeutic_class"]
        member = member_profiles.get(patient_id, {})
        barriers = barriers_by_class.get((patient_id, therapeutic_class), [])
        med = medication_metadata.get((patient_id, therapeutic_class), {})

        for intervention in intervention_catalog:
            # Hard eligibility filters. Each one is a documented
            # business rule; comments explain why.

            # Brand-only interventions only apply to brand-name fills.
            if intervention["brand_only"] and not med.get("is_brand", False):
                continue

            # Class-specific interventions (e.g., statin education
            # module) only apply to the matching class.
            applicable_classes = intervention.get("applicable_classes")
            if applicable_classes and therapeutic_class not in applicable_classes:
                continue

            # LIS-only interventions (e.g., LIS enrollment assistance)
            # only apply to LIS-eligible members.
            if intervention["requires_lis"] and not member.get("lis_enrolled", False):
                continue

            # Med-sync requires the member's preferred pharmacy to be
            # a partner pharmacy. Without partnership, the API path
            # doesn't exist; fall back to a pharmacist referral
            # instead. The orchestrator handles that fallback.
            if (intervention["requires_partner_pharmacy"]
                and member.get("preferred_pharmacy_id") not in PARTNER_PHARMACIES):
                continue

            # Language eligibility: a Spanish-only patient won't be
            # served by an English-only intervention.
            preferred_lang = member.get("preferred_language", "en")
            if preferred_lang not in intervention.get("supported_languages", ["en"]):
                continue

            # Cooldown: don't re-recommend the same intervention type
            # if the member already received one recently.
            cooldown = intervention.get("cooldown_days", 0)
            if cooldown > 0 and _within_cooldown(member, intervention, cooldown):
                continue

            # Mutual-exclusion: if a higher-touch intervention is
            # already in flight for this (patient, class), don't queue
            # a competing one. The check here is a placeholder; the
            # production version queries the engagement-events table
            # for in-flight interventions per (patient, class).
            if _has_inflight_conflicting(patient_id, therapeutic_class, intervention):
                continue

            candidates.append({
                "patient_id":              patient_id,
                "therapeutic_class":       therapeutic_class,
                "intervention_id":         intervention["intervention_id"],
                "intervention_type":       intervention["type"],
                "intervention_cost":       intervention["marginal_cost"],
                "is_high_touch":           intervention["is_high_touch"],
                "supported_barriers":      intervention["supported_barriers"],
                "ranked_barriers":         barriers,
                "data_quality_flag":       med.get("data_quality_flag", "complete"),
                "run_date":                run_date,
            })

    logger.info(
        "Built %d (patient, intervention, class) candidate triples", len(candidates),
    )
    return candidates

def _within_cooldown(member: dict, intervention: dict, cooldown_days: int) -> bool:
    """
    True if the member already received this intervention type within
    `cooldown_days`. The patient-profile table maintains a
    `prior_interventions` map keyed on intervention_type with the
    most-recent completion timestamp.
    """
    prior = member.get("prior_interventions", {}).get(intervention["type"])
    if not prior:
        return False
    last_completed = prior.get("completed_at")
    if not last_completed:
        return False
    try:
        last_dt = datetime.datetime.fromisoformat(last_completed)
    except ValueError:
        return False
    delta = datetime.datetime.now(timezone.utc) - last_dt.astimezone(timezone.utc)
    return delta.days < cooldown_days

def _has_inflight_conflicting(patient_id: str, therapeutic_class: str,
                               intervention: dict) -> bool:
    """
    Placeholder for the in-flight conflict check. Production: query
    the recommendation-log for (patient, class) rows in the last
    N days where intervention_status is 'queued', 'sent', or
    'engaged-not-completed'. The example returns False so the demo
    proceeds.
    """
    return False

def score_candidates(candidates: list, run_date: str) -> list:
    """
    Score the candidate triples on need, barrier-fit, engagement, and
    uplift. Returns the same candidates with score columns appended.

    The example uses synthetic scoring so the demo runs offline.
    Production: write candidates to S3 by intervention_id, kick off
    Batch Transform jobs in parallel, wait, read scores back.
    """
    # ---- Group candidates by intervention_id ----
    by_intervention = {}
    for c in candidates:
        by_intervention.setdefault(c["intervention_id"], []).append(c)

    # ---- Production-shape: kick off Batch Transform jobs ----
    # The example calls a synthetic scorer that returns deterministic
    # values; production replaces _score_candidates_via_batch_transform
    # with the real submit-and-wait logic.
    scored = []
    for intervention_id, triples in by_intervention.items():
        intervention_scores = _score_candidates_via_batch_transform(
            triples, intervention_id, run_date,
        )
        scored.extend(intervention_scores)

    # ---- Barrier-fit is a deterministic computation ----
    for c in scored:
        c["barrier_fit"] = _compute_barrier_fit(
            c["ranked_barriers"], c["supported_barriers"],
        )

    return scored

def _score_candidates_via_batch_transform(
    triples: list, intervention_id: str, run_date: str,
) -> list:
    """
    Production: write candidates to S3, kick off three Batch Transform
    jobs (need / engagement / uplift) for this intervention, wait,
    join the scores back to the candidates.

    The example uses a synthetic scorer that derives plausible scores
    from observed adherence + barrier signal. Sufficient for the
    pipeline demo; nowhere near production accuracy.
    """
    scored = []
    for c in triples:
        # Need score: heuristic from PDC and class. Production reads
        # this from the per-class need model output.
        pdc = _lookup_pdc(c["patient_id"], c["therapeutic_class"])
        if pdc < 0.50:
            need = 0.85
        elif pdc < 0.65:
            need = 0.65
        elif pdc < 0.80:
            need = 0.45
        else:
            need = 0.20

        # Engagement probability: heuristic from intervention type and
        # member engagement quartile. Production reads this from the
        # per-intervention engagement model.
        engagement_quartile = _lookup_engagement_quartile(c["patient_id"])
        type_base = {
            "text_reminder":          0.30,
            "education":              0.18,
            "pharmacist_consult":     0.42,
            "cost_assistance":        0.50,
            "med_sync":               0.55,
            "regimen_simplification": 0.35,
        }.get(c["intervention_type"], 0.30)
        quartile_adj = {"q1": -0.10, "q2": -0.05, "q3": 0.0, "q4": 0.10}.get(
            engagement_quartile, 0.0
        )
        engagement_prob = max(0.05, min(0.95, type_base + quartile_adj))

        # Uplift estimate: heuristic from barrier-fit-with-rule-top
        # and intervention type. Production reads this from the
        # per-intervention uplift model (causal forest / X-learner).
        top_barrier = c["ranked_barriers"][0]["barrier"] if c["ranked_barriers"] else None
        type_uplift_base = {
            "text_reminder":          0.05,
            "education":              0.07,
            "pharmacist_consult":     0.18,
            "cost_assistance":        0.20,
            "med_sync":               0.12,
            "regimen_simplification": 0.10,
        }.get(c["intervention_type"], 0.05)
        # If the intervention's top supported barrier matches the
        # patient's top barrier, bump the uplift estimate.
        supported = c["supported_barriers"]
        barrier_match_bonus = 0.0
        if top_barrier and supported.get(top_barrier, 0.0) >= 0.7:
            barrier_match_bonus = 0.10
        uplift = max(0.0, type_uplift_base + barrier_match_bonus)

        scored.append({
            **c,
            "need_score":         need,
            "engagement_prob":    engagement_prob,
            "uplift_estimate":    uplift,
        })
    return scored

def _compute_barrier_fit(ranked_barriers: list, supported_barriers: dict) -> float:
    """
    Dot product between the patient's ranked-barriers vector and the
    intervention's supported-barriers vector.

    Example: patient barriers [cost: 0.72, beliefs: 0.21] dotted with
    cost-assistance supports {cost: 1.0, beliefs: 0.0, ...} = 0.72.
    Same patient with text-reminder supports {forgetfulness: 1.0,
    cost: 0.0, ...} = 0.0.
    """
    score = 0.0
    for entry in ranked_barriers:
        b = entry["barrier"]
        score += entry["confidence"] * supported_barriers.get(b, 0.0)
    return round(score, 4)

def _lookup_pdc(patient_id: str, therapeutic_class: str) -> float:
    """Placeholder lookup. Production: read from Feature Store online store."""
    return _DEMO_PDC_LOOKUP.get((patient_id, therapeutic_class), 0.75)

def _lookup_engagement_quartile(patient_id: str) -> str:
    """Placeholder lookup. Production: read from patient-profile table."""
    return _DEMO_ENGAGEMENT_QUARTILE.get(patient_id, "q3")

# Demo state populated by the runner at the bottom of this file. The
# production functions above replace these globals with real lookups.
_DEMO_PDC_LOOKUP: dict = {}
_DEMO_ENGAGEMENT_QUARTILE: dict = {}

def compute_priority(
    scored_candidates: list,
    policy_weights: dict = POLICY_WEIGHTS,
    policy_version: str = POLICY_VERSION,
) -> list:
    """
    Compute combined priority per (patient, intervention, medication)
    triple, including the per-component contributions for downstream
    auditing.

    Normalization choice: min-max within intervention_type for
    engagement and uplift (so a high engagement-prob for a reminder
    is comparable to a high engagement-prob for a pharmacist consult);
    raw values for need (which is normalized within the per-class
    model already); raw values for barrier-fit (already in [0, 1]).

    Returns the candidates with priority and priority_components
    fields appended.
    """
    if not scored_candidates:
        return []

    # ---- Normalize engagement and uplift within intervention_type ----
    by_type: dict = {}
    for c in scored_candidates:
        by_type.setdefault(c["intervention_type"], []).append(c)

    for intervention_type, type_rows in by_type.items():
        for component in ("engagement_prob", "uplift_estimate"):
            values = [r[component] for r in type_rows]
            lo, hi = min(values), max(values)
            spread = hi - lo if hi > lo else 1.0
            for r in type_rows:
                r[f"{component}_norm"] = (r[component] - lo) / spread

        # Cost-efficiency = uplift / intervention_cost, normalized
        # within intervention type so the $0.05 reminder competes
        # with the $55 pharmacist consult on a like-for-like axis.
        ce_values = []
        for r in type_rows:
            ce = r["uplift_estimate"] / max(r["intervention_cost"], MIN_COST_FOR_DIVISION)
            r["_cost_efficiency_raw"] = ce
            ce_values.append(ce)
        ce_lo, ce_hi = min(ce_values), max(ce_values)
        ce_spread = ce_hi - ce_lo if ce_hi > ce_lo else 1.0
        for r in type_rows:
            r["cost_efficiency_norm"] = (r["_cost_efficiency_raw"] - ce_lo) / ce_spread

    # ---- Combine into priority ----
    for c in scored_candidates:
        priority_components = {
            "need_contrib":            policy_weights["need"]            * c["need_score"],
            "barrier_fit_contrib":     policy_weights["barrier_fit"]     * c["barrier_fit"],
            "engagement_contrib":      policy_weights["engagement"]      * c["engagement_prob_norm"],
            "uplift_contrib":          policy_weights["uplift"]          * c["uplift_estimate_norm"],
            "cost_efficiency_contrib": policy_weights["cost_efficiency"] * c["cost_efficiency_norm"],
        }
        c["priority"] = round(sum(priority_components.values()), 4)
        c["priority_components"] = {
            k: round(v, 4) for k, v in priority_components.items()
        }
        c["policy_version"] = policy_version

    # ---- Group by (patient, class) and rank within ----
    by_patient_class: dict = {}
    for c in scored_candidates:
        key = (c["patient_id"], c["therapeutic_class"])
        by_patient_class.setdefault(key, []).append(c)

    for key, rows in by_patient_class.items():
        rows.sort(key=lambda x: x["priority"], reverse=True)
        for rank_pos, r in enumerate(rows, start=1):
            r["intervention_rank_within_medication"] = rank_pos

    logger.info(
        "Computed priority for %d candidates across %d (patient, class) pairs",
        len(scored_candidates), len(by_patient_class),
    )
    return scored_candidates

def allocate_heterogeneous(
    prioritized: list,
    intervention_catalog: list,
    member_profiles: dict,
    policy_weights: dict = POLICY_WEIGHTS,
    equity_floors: dict = EQUITY_FLOORS,
    run_horizon_days: int = 7,
    run_date: str = None,
) -> list:
    """
    Greedy allocation with multi-intervention-per-patient and equity
    floors.

    Returns a list of allocation records with priority, components,
    cohort features, and allocation_reason. Persists to the
    recommendation-log table.
    """
    run_date = run_date or _today_str()
    if not prioritized:
        return []

    candidates_sorted = sorted(prioritized, key=lambda x: x["priority"], reverse=True)

    intervention_by_id = {i["intervention_id"]: i for i in intervention_catalog}

    # ---- Initialize per-intervention capacity counters ----
    capacity_remaining = {
        i["intervention_id"]: i["daily_capacity"] * run_horizon_days
        for i in intervention_catalog
    }

    # ---- Initialize per-intervention equity-floor counters ----
    equity_remaining = {
        i["intervention_id"]: dict(equity_floors.get(i["intervention_id"], {}))
        for i in intervention_catalog
    }

    # ---- Per-patient counters ----
    patient_intervention_count: dict = {}
    patient_high_touch_count: dict = {}
    patient_contact_count_30d: dict = {}

    allocated = []
    allocated_by_patient_class: dict = {}    # for cross-intervention exclusion

    # ---- Greedy primary pass ----
    for candidate in candidates_sorted:
        patient_id = candidate["patient_id"]
        therapeutic_class = candidate["therapeutic_class"]
        intervention_id = candidate["intervention_id"]
        intervention = intervention_by_id[intervention_id]
        member = member_profiles.get(patient_id, {})

        # Per-intervention capacity.
        if capacity_remaining.get(intervention_id, 0) <= 0:
            continue

        # Per-patient interventions-per-run cap.
        if (patient_intervention_count.get(patient_id, 0)
            >= MAX_INTERVENTIONS_PER_PATIENT_PER_RUN):
            continue

        # Per-patient high-touch cap.
        if intervention["is_high_touch"]:
            if (patient_high_touch_count.get(patient_id, 0)
                >= MAX_HIGH_TOUCH_PER_PATIENT_PER_RUN):
                continue

        # Per-patient 30-day contact cap. Sum existing contacts plus
        # contacts queued in this run.
        existing_contacts = int(member.get("outreach_recent_30d_count", 0))
        new_contacts_this_run = patient_contact_count_30d.get(patient_id, 0)
        if (intervention["generates_patient_contact"]
            and (existing_contacts + new_contacts_this_run)
                >= MAX_CONTACTS_PER_PATIENT_30D):
            continue

        # Cross-intervention exclusion: a pharmacist consult on a
        # medication suppresses a reminder on the same medication
        # (the pharmacist will handle the reminder framing).
        already = allocated_by_patient_class.get((patient_id, therapeutic_class), [])
        if _conflicts_with(already, intervention):
            continue

        # ---- Equity-floor accounting ----
        cohort_features = _lookup_cohort_features_from_profile(member)
        applicable = _applicable_floor_cohorts(
            cohort_features, equity_floors.get(intervention_id, {})
        )
        used_floor = None
        for floor_cohort in applicable:
            if equity_remaining[intervention_id].get(floor_cohort, 0) > 0:
                equity_remaining[intervention_id][floor_cohort] -= 1
                used_floor = floor_cohort
                break

        # ---- Commit the allocation ----
        capacity_remaining[intervention_id] -= 1
        patient_intervention_count[patient_id] = patient_intervention_count.get(patient_id, 0) + 1
        if intervention["is_high_touch"]:
            patient_high_touch_count[patient_id] = patient_high_touch_count.get(patient_id, 0) + 1
        if intervention["generates_patient_contact"]:
            patient_contact_count_30d[patient_id] = patient_contact_count_30d.get(patient_id, 0) + 1

        allocation_reason = (
            f"equity_floor:{used_floor}" if used_floor
            else "top_priority_general_capacity"
        )

        record = {
            "tracking_id": _make_tracking_id(
                run_date, patient_id, therapeutic_class, intervention_id,
            ),
            "run_date":              run_date,
            "patient_id":            patient_id,
            "therapeutic_class":     therapeutic_class,
            "intervention_id":       intervention_id,
            "intervention_type":     intervention["type"],
            "priority":              candidate["priority"],
            "priority_components":   candidate["priority_components"],
            "policy_version":        candidate["policy_version"],
            "cohort_features":       cohort_features,
            "allocation_reason":     allocation_reason,
            "data_quality_flag":     candidate.get("data_quality_flag", "complete"),
            "ranked_barriers":       candidate.get("ranked_barriers", []),
        }
        allocated.append(record)
        allocated_by_patient_class.setdefault(
            (patient_id, therapeutic_class), []
        ).append(intervention)

    # ---- Equity-floor top-up pass ----
    # If any floor wasn't filled in the primary pass, pull additional
    # candidates from the cohort to fill the reserved slots.
    for intervention_id, floors in equity_remaining.items():
        for floor_cohort, remaining in list(floors.items()):
            if remaining <= 0:
                continue
            logger.info(
                "Top-up pass: intervention=%s floor=%s remaining=%d",
                intervention_id, floor_cohort, remaining,
            )
            cohort_pool = []
            for c in candidates_sorted:
                if c["intervention_id"] != intervention_id:
                    continue
                pid = c["patient_id"]
                if patient_intervention_count.get(pid, 0) >= MAX_INTERVENTIONS_PER_PATIENT_PER_RUN:
                    continue
                if capacity_remaining[intervention_id] <= 0:
                    break
                cohort_features = _lookup_cohort_features_from_profile(
                    member_profiles.get(pid, {}),
                )
                if floor_cohort in _applicable_floor_cohorts(
                    cohort_features, {floor_cohort: 1},
                ):
                    cohort_pool.append((c, cohort_features))

            for c, cohort_features in cohort_pool[:remaining]:
                if capacity_remaining[intervention_id] <= 0:
                    break
                pid = c["patient_id"]
                tc = c["therapeutic_class"]
                intervention = intervention_by_id[intervention_id]

                # Re-check the per-patient and contact caps for the
                # top-up candidate.
                if patient_intervention_count.get(pid, 0) >= MAX_INTERVENTIONS_PER_PATIENT_PER_RUN:
                    continue
                if intervention["is_high_touch"] and (
                    patient_high_touch_count.get(pid, 0) >= MAX_HIGH_TOUCH_PER_PATIENT_PER_RUN
                ):
                    continue

                capacity_remaining[intervention_id] -= 1
                patient_intervention_count[pid] = patient_intervention_count.get(pid, 0) + 1
                if intervention["is_high_touch"]:
                    patient_high_touch_count[pid] = patient_high_touch_count.get(pid, 0) + 1
                if intervention["generates_patient_contact"]:
                    patient_contact_count_30d[pid] = patient_contact_count_30d.get(pid, 0) + 1
                equity_remaining[intervention_id][floor_cohort] -= 1

                allocated.append({
                    "tracking_id": _make_tracking_id(
                        run_date, pid, tc, intervention_id,
                    ),
                    "run_date":              run_date,
                    "patient_id":            pid,
                    "therapeutic_class":     tc,
                    "intervention_id":       intervention_id,
                    "intervention_type":     intervention["type"],
                    "priority":              c["priority"],
                    "priority_components":   c["priority_components"],
                    "policy_version":        c["policy_version"],
                    "cohort_features":       cohort_features,
                    "allocation_reason":     f"equity_floor_topup:{floor_cohort}",
                    "data_quality_flag":     c.get("data_quality_flag", "complete"),
                    "ranked_barriers":       c.get("ranked_barriers", []),
                })

    # ---- Persist to the recommendation-log table ----
    rec_table = dynamodb.Table(RECOMMENDATION_LOG_TABLE)
    try:
        with rec_table.batch_writer() as batch:
            for row in allocated:
                batch.put_item(Item=_to_decimal_dict({
                    **row,
                    "created_at": _now_iso(),
                }))
    except Exception as exc:
        logger.warning("Recommendation-log batch write failed: %s", exc)

    _emit_metric(
        "adherence_allocations_made",
        value=len(allocated),
        dimensions={"run_date": run_date, "policy_version": POLICY_VERSION},
    )
    logger.info(
        "Allocated %d (patient, intervention, class) triples across %d patients",
        len(allocated), len(patient_intervention_count),
    )
    return allocated

def _conflicts_with(already_allocated: list, intervention: dict) -> bool:
    """
    Cross-intervention exclusion logic. The full table lives in the
    intervention-catalog config; the example codes the most common
    case: a pharmacist consult absorbs lower-touch interventions for
    the same medication.
    """
    types_already = {i["type"] for i in already_allocated}
    if "pharmacist_consult" in types_already:
        # Don't queue a reminder or education on a class where a
        # pharmacist consult is already in flight.
        if intervention["type"] in {"text_reminder", "education"}:
            return True
    return False

def _lookup_cohort_features_from_profile(member: dict) -> dict:
    """Pull cohort features from the patient-profile dict."""
    return {
        "engagement_history_quartile": member.get("engagement_history_quartile", "q3"),
        "language":                    member.get("preferred_language", "en"),
        "sdoh_cohort":                 member.get("sdoh_cohort"),
        "age_band":                    member.get("age_band"),
    }

def _applicable_floor_cohorts(cohort_features: dict, floor_definitions: dict) -> list:
    """Return the floor names the candidate qualifies for."""
    result = []
    for floor_name in floor_definitions:
        if floor_name == "engagement_q1" and cohort_features.get("engagement_history_quartile") == "q1":
            result.append(floor_name)
        elif floor_name == "language_non_en" and cohort_features.get("language") not in (None, "en"):
            result.append(floor_name)
        elif floor_name == "sdoh_low_food_security" and cohort_features.get("sdoh_cohort") == "low_food_security":
            result.append(floor_name)
    return result

def orchestrate_interventions(
    allocated: list,
    member_profiles: dict,
    medication_metadata: dict,
    run_date: str,
) -> list:
    """
    For each allocated triple, dispatch through the appropriate
    channel: SMS/email reminder, education content match, pharmacist
    queue with pre-call brief, cost-assistance staff queue, partner-
    pharmacy med-sync API, or PCP care-team inbox.

    Returns a list of dispatch records for audit and downstream
    reconciliation.
    """
    profile_table = dynamodb.Table(PATIENT_PROFILE_TABLE)
    dispatched = []

    for row in allocated:
        member = member_profiles.get(row["patient_id"], {})
        med = medication_metadata.get(
            (row["patient_id"], row["therapeutic_class"]), {}
        )

        # Build the de-identified context the LLM tasks consume.
        # Identifiers are NOT in this dict; the channel optimizer
        # re-attaches them after generation.
        de_id_context = {
            "therapeutic_class":      row["therapeutic_class"],
            "adherence_summary":      _summarize_adherence_for_outreach(
                row["patient_id"], row["therapeutic_class"],
            ),
            "ranked_barriers":        row.get("ranked_barriers", []),
            "preferred_language":     member.get("preferred_language", "en"),
            "tone":                   "supportive, non-alarming",
            "data_quality_flag":      row.get("data_quality_flag", "complete"),
        }

        try:
            if row["intervention_type"] == "text_reminder":
                dispatch = _dispatch_text_reminder(row, member, de_id_context)
            elif row["intervention_type"] == "education":
                dispatch = _dispatch_education(row, member, de_id_context)
            elif row["intervention_type"] == "pharmacist_consult":
                dispatch = _dispatch_pharmacist_consult(row, member, med, de_id_context)
            elif row["intervention_type"] == "cost_assistance":
                dispatch = _dispatch_cost_assistance(row, member, med)
            elif row["intervention_type"] == "med_sync":
                dispatch = _dispatch_med_sync(row, member)
            elif row["intervention_type"] == "regimen_simplification":
                dispatch = _dispatch_regimen_simplification(row, member, med, de_id_context)
            else:
                logger.warning(
                    "Unknown intervention_type=%s; skipping", row["intervention_type"],
                )
                continue
            dispatched.append(dispatch)
        except Exception as exc:
            logger.warning(
                "Dispatch failed for tracking_id=%s: %s", row["tracking_id"], exc,
            )
            continue

        # Update the contact-cap counter optimistically when patient
        # contact is generated. Reconcile in the engagement-attribution
        # step on outreach-failed events.
        if _intervention_generates_contact(row["intervention_type"]):
            try:
                profile_table.update_item(
                    Key={"patient_id": row["patient_id"]},
                    UpdateExpression=(
                        "ADD outreach_recent_30d_count :one "
                        "SET outreach_last_at = :now"
                    ),
                    ExpressionAttributeValues={
                        ":one": Decimal("1"),
                        ":now": _now_iso(),
                    },
                )
            except Exception as exc:
                logger.warning(
                    "Failed to update contact counter for %s: %s",
                    row["patient_id"], exc,
                )

        # Emit the adherence_intervention_recommended event. This is
        # the join point for downstream attribution.
        try:
            kinesis_client.put_record(
                StreamName=ENGAGEMENT_STREAM_NAME,
                PartitionKey=row["patient_id"],
                Data=json.dumps({
                    "event_type":          "adherence_intervention_recommended",
                    "tracking_id":         row["tracking_id"],
                    "patient_id":          row["patient_id"],
                    "therapeutic_class":   row["therapeutic_class"],
                    "intervention_id":     row["intervention_id"],
                    "intervention_type":   row["intervention_type"],
                    "run_date":            row["run_date"],
                    "priority_components": row["priority_components"],
                    "allocation_reason":   row["allocation_reason"],
                    "timestamp":           _now_iso(),
                }, default=str).encode("utf-8"),
            )
        except Exception as exc:
            logger.warning(
                "Failed to publish recommended event for %s: %s",
                row["tracking_id"], exc,
            )

    logger.info("Dispatched %d intervention assignments", len(dispatched))
    return dispatched

def _intervention_generates_contact(intervention_type: str) -> bool:
    """Whether this intervention type produces a patient-facing contact."""
    return intervention_type in {
        "text_reminder", "education", "pharmacist_consult",
        "cost_assistance", "med_sync",
    }

def _summarize_adherence_for_outreach(patient_id: str, therapeutic_class: str) -> dict:
    """
    Build a compact, high-level adherence summary for prompt context.
    Stays at the tier of "mostly on schedule, a few gaps" rather than
    surfacing exact PDC values that might leak into member-facing
    copy via LLM hallucination.
    """
    pdc = _lookup_pdc(patient_id, therapeutic_class)
    if pdc >= 0.80:
        return {"status": "mostly_on_schedule"}
    if pdc >= 0.50:
        return {"status": "occasional_gaps"}
    return {"status": "frequent_gaps"}

def _dispatch_text_reminder(row: dict, member: dict, ctx: dict) -> dict:
    """
    Generate a tailored SMS/email reminder via Bedrock and hand it
    to the channel optimizer. Falls back to the default template on
    LLM or validator failure.
    """
    intervention_id = row["intervention_id"]
    catalog_entry = _lookup_intervention(intervention_id)
    tailored = None
    try:
        tailored = _bedrock_tailor_reminder(catalog_entry, ctx)
        if not _validate_reminder(tailored):
            logger.warning(
                "Reminder validation failed for %s; falling back to template",
                row["tracking_id"],
            )
            tailored = None
    except Exception as exc:
        logger.warning("Reminder tailoring failed for %s: %s", row["tracking_id"], exc)

    return _queue_via_channel_optimizer({
        "tracking_id":       row["tracking_id"],
        "patient_id":        row["patient_id"],
        "therapeutic_class": row["therapeutic_class"],
        "content_type":      "adherence_reminder",
        "tailored":          tailored,
        "fallback_template": catalog_entry["default_template"],
        "urgency":           "standard",
        "queued_at":         _now_iso(),
    })

def _dispatch_education(row: dict, member: dict, ctx: dict) -> dict:
    """
    Match an education content artifact to the patient's barrier.
    Production: Recipe 4.2's content-matching pipeline. The example
    returns a placeholder match.
    """
    catalog_entry = _lookup_intervention(row["intervention_id"])
    return _queue_via_channel_optimizer({
        "tracking_id":       row["tracking_id"],
        "patient_id":        row["patient_id"],
        "therapeutic_class": row["therapeutic_class"],
        "content_type":      "adherence_education",
        "tailored":          None,
        "fallback_template": catalog_entry["default_template"],
        "urgency":           "standard",
        "queued_at":         _now_iso(),
    })

def _dispatch_pharmacist_consult(
    row: dict, member: dict, med: dict, ctx: dict,
) -> dict:
    """
    Generate the pharmacist pre-call brief and enqueue the consult.
    The brief is the structured pre-work the pharmacist reads before
    dialing; saves 5-10 minutes per call and improves quality.
    """
    try:
        brief = _bedrock_pharmacist_brief(member, med, ctx)
        if not _validate_pharmacist_brief(brief):
            brief = {"summary": "(brief generation failed; pharmacist to review chart manually)"}
    except Exception as exc:
        logger.warning("Pharmacist brief failed for %s: %s", row["tracking_id"], exc)
        brief = {"summary": "(brief generation failed; pharmacist to review chart manually)"}

    enqueue_record = {
        "tracking_id":         row["tracking_id"],
        "patient_id":          row["patient_id"],
        "therapeutic_class":   row["therapeutic_class"],
        "priority":            row["priority"],
        "suspected_barrier":   ctx["ranked_barriers"][0] if ctx["ranked_barriers"] else None,
        "brief":               brief,
        "target_window":       _compute_target_window(),
        "queued_at":           _now_iso(),
    }
    logger.info(
        "Queued pharmacist consult tracking_id=%s (priority=%.3f)",
        row["tracking_id"], row["priority"],
    )
    return enqueue_record

def _dispatch_cost_assistance(row: dict, member: dict, med: dict) -> dict:
    """
    Hand to the case-management queue. The case manager works through
    the cost-assistance cascade (formulary substitution, manufacturer
    copay card, foundation grants, LIS, state programs, generic
    alternatives) and writes outcomes back via engagement events.
    """
    enqueue_record = {
        "tracking_id":         row["tracking_id"],
        "patient_id":          row["patient_id"],
        "therapeutic_class":   row["therapeutic_class"],
        "medication_ndc":      med.get("ndc"),
        "suspected_barrier":   "cost",
        "member_lis_status":   bool(member.get("lis_enrolled", False)),
        "cost_sharing_tier":   med.get("formulary_tier"),
        "queued_at":           _now_iso(),
    }
    logger.info("Queued cost-assistance tracking_id=%s", row["tracking_id"])
    return enqueue_record

def _dispatch_med_sync(row: dict, member: dict) -> dict:
    """
    Route to the partner pharmacy API if available, otherwise to a
    pharmacist-mediated med-sync conversation queue.
    """
    pharmacy_id = member.get("preferred_pharmacy_id")
    if pharmacy_id in PARTNER_PHARMACIES:
        record = {
            "tracking_id":   row["tracking_id"],
            "patient_id":    row["patient_id"],
            "pharmacy_id":   pharmacy_id,
            "channel":       "partner_pharmacy_api",
            "queued_at":     _now_iso(),
        }
        logger.info("Queued med-sync via partner API for %s", row["tracking_id"])
        return record
    record = {
        "tracking_id":         row["tracking_id"],
        "patient_id":          row["patient_id"],
        "therapeutic_class":   row["therapeutic_class"],
        "channel":             "pharmacist_referral_queue",
        "queued_at":           _now_iso(),
    }
    logger.info(
        "Queued med-sync via pharmacist referral for %s (no partner pharmacy)",
        row["tracking_id"],
    )
    return record

def _dispatch_regimen_simplification(
    row: dict, member: dict, med: dict, ctx: dict,
) -> dict:
    """
    Build a PCP briefing and post to the EHR care-team inbox. The PCP
    decides whether to act (combination pill, once-daily, blister
    pack) at the next visit; a pcp_override event flows back if they
    decline.
    """
    try:
        briefing = _bedrock_pcp_briefing(member, med, ctx)
    except Exception as exc:
        logger.warning("PCP briefing failed for %s: %s", row["tracking_id"], exc)
        briefing = {
            "talking_points": ["Consider regimen simplification for adherence"],
            "rationale":      "Briefing generation failed; review chart",
        }

    record = {
        "tracking_id":     row["tracking_id"],
        "patient_id":      row["patient_id"],
        "channel":         "ehr_care_team_inbox",
        "briefing":        briefing,
        "suggested_action": "consider regimen simplification",
        "queued_at":       _now_iso(),
    }
    logger.info("Posted PCP briefing for %s", row["tracking_id"])
    return record

def _queue_via_channel_optimizer(payload: dict) -> dict:
    """
    Hand to Recipe 4.1's channel optimizer. The example logs and
    returns; production posts to an SQS queue or invokes the channel
    optimizer Lambda directly.
    """
    logger.info(
        "Queued via channel optimizer: tracking_id=%s urgency=%s",
        payload["tracking_id"], payload.get("urgency"),
    )
    return payload

def _compute_target_window() -> dict:
    """
    The window in which the pharmacist should attempt the consult.
    Tighter window for time-sensitive interventions; this example
    uses a 7-day default.
    """
    today = datetime.date.today()
    return {
        "start": today.isoformat(),
        "end":   (today + datetime.timedelta(days=7)).isoformat(),
    }

def _lookup_intervention(intervention_id: str) -> dict:
    """Fetch an intervention record from the catalog. Production: DynamoDB lookup."""
    for i in SAMPLE_INTERVENTIONS:
        if i["intervention_id"] == intervention_id:
            return i
    raise KeyError(f"Unknown intervention_id: {intervention_id}")