Recipe 4.8: Python Implementation Example

pip install boto3 pandas numpy

import json
import logging
import math
import re
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, treatment_id,
# comparator_id, predicted_effect, similar_patient_cohort_features)
# join. The row implicitly identifies the patient, the suspected
# clinical condition, and the treatment options being weighed; the
# scoring-results, briefings, and decision-records tables are
# clinical-record-equivalent PHI.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from SageMaker, DynamoDB, Bedrock,
# Kinesis, S3, and Athena. The point-of-care scoring path is bursty
# (clinician opens a chart and requests scoring), and the weekly
# retraining and monthly surveillance pipelines run heavy batch loads.
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)

# --- Bedrock Model Configuration ---
# Three distinct LLM use cases. Clinician-facing comparison briefings
# go to a Sonnet-class model because the prompt is long-context and
# the validator's recommendation-language rule is strict; the larger
# model gives a better first-pass-pass rate. Patient-facing summaries
# and disagreement-investigation narratives use a Haiku-class model
# because they're shorter and cheaper at scale.
COMPARISON_BRIEFING_MODEL_ID    = "anthropic.claude-3-5-sonnet-20241022-v2:0"
PATIENT_SUMMARY_MODEL_ID         = "anthropic.claude-3-5-haiku-20241022-v1:0"
DISAGREEMENT_NARRATIVE_MODEL_ID  = "anthropic.claude-3-5-haiku-20241022-v1:0"

# --- DynamoDB Table Names ---
# Nine tables. Keep them separate so access patterns stay clean and
# IAM scoping is precise:
#   1. treatment-catalog:           catalog of treatments and metadata
#   2. treatment-comparison-pairs:  per-pair model and protocol pointers
#   3. cohort-metadata:             per-(pair, protocol_version, run)
#                                    archive pointer
#   4. scoring-results:             per (patient, scoring_run)
#                                    structured result feeding briefings
#   5. briefings:                   LLM-generated clinician briefings
#   6. decision-records:            clinician's recorded decision plus
#                                    frozen-at-decision-time predictions
#   7. prediction-outcome-pairs:    matched predicted-vs-actual outcomes
#   8. governance-review-tasks:     pending-promotion review queue
#   9. surveillance-alerts:         calibration drift, fairness drift,
#                                    adverse-event rate alerts
TREATMENT_CATALOG_TABLE          = "treatment-catalog"
TREATMENT_COMPARISON_PAIRS_TABLE = "treatment-comparison-pairs"
COHORT_METADATA_TABLE            = "cohort-metadata"
SCORING_RESULTS_TABLE            = "scoring-results"
BRIEFINGS_TABLE                  = "briefings"
DECISION_RECORDS_TABLE           = "decision-records"
PREDICTION_OUTCOME_PAIRS_TABLE   = "prediction-outcome-pairs"
GOVERNANCE_REVIEW_TASKS_TABLE    = "governance-review-tasks"
SURVEILLANCE_ALERTS_TABLE        = "surveillance-alerts"

# --- S3 Buckets ---
# Production: each bucket has its own KMS key and bucket policy.
# Replace placeholder names with your account's buckets.
TRX_COHORTS_BUCKET       = "trx-cohorts"
TRX_EVAL_BUCKET          = "trx-eval"
TRX_SURVEILLANCE_BUCKET  = "trx-surveillance"
TRX_DATA_LAKE_BUCKET     = "trx-data-lake"
ATHENA_RESULTS_BUCKET    = "trx-athena-results"

# --- Athena ---
ATHENA_WORKGROUP = "trx-recommender"
ATHENA_DATABASE  = "trx_data_lake"

# --- Kinesis ---
# Same engagement-event-bus pattern as Recipes 4.4 through 4.7, with
# new event types specific to this recipe: cohort_constructed,
# training_suspended, governance_review_pending, model_promoted,
# treatment_scoring_requested, treatment_scoring_completed,
# treatment_scoring_oodflag_triggered, briefing_generated,
# briefing_validator_fallback, treatment_decision_recorded,
# treatment_outcome_observed, prediction_calibration_alert,
# disagreement_alert.
TRX_EVENTS_STREAM_NAME = "trx-events"

# --- Run Configuration ---
POLICY_VERSION = "trx-policy-v0.1"

# CATE ensemble configuration. The example simulates an ensemble of
# three estimators (causal forest, DR-learner, BART). Production
# trains all three through SageMaker BYOC containers with EconML,
# bartCause, or grf wrappers and registers them in the SageMaker
# Model Registry. The agreement threshold below determines when
# disagreement triggers an alert; lower values are stricter.
CATE_ESTIMATOR_NAMES = ["causal_forest", "dr_learner", "bart"]
DISAGREEMENT_THRESHOLD = 0.30   # spread of point estimates across
                                # estimators normalized to the
                                # outcome's natural scale

# Minimum sample size for fairness analysis within a protected
# cohort. Below this size, fairness metrics are too unreliable to
# act on; the system reports "insufficient sample" rather than a
# misleading number.
MIN_FAIRNESS_SAMPLE_SIZE = 30

# Calibration-slope thresholds. A well-calibrated model has slope
# near 1.0; values outside this band trigger a calibration warning
# at the governance gate.
CALIBRATION_SLOPE_LOW   = 0.7
CALIBRATION_SLOPE_HIGH  = 1.3

# Surveillance window for calibration drift detection (days).
# Production typically uses 90 days for outcomes that read out at
# 90 days and longer for outcomes with longer time-to-readout.
SURVEILLANCE_WINDOW_DAYS  = 90
DRIFT_ALERT_THRESHOLD     = 0.20  # absolute change in slope or
                                  # intercept relative to baseline
COHORT_DRIFT_ALERT_THRESHOLD = 0.25

# OOD policy. The example computes a simple severity from propensity
# distance and density of similar patients; production uses a more
# sophisticated combination of propensity tail-flag, embedding-
# distance, and isolation-forest-based anomaly score.
OOD_SEVERITY_WARNING_THRESHOLD = 0.50
OOD_SEVERITY_SUPPRESS_THRESHOLD = 0.85

# Minimum size of similar-patient cohort to summarize. Below this
# size, the demographic and outcome summaries are too noisy to be
# useful; the system suppresses the cohort summary rather than
# present a misleading number.
MIN_SIMILAR_PATIENT_COHORT = 25

# Briefing TTL. A briefing generated for a Tuesday visit should not
# be presented unchanged at a Friday visit; the EHR re-requests
# scoring if the briefing is older than this.
BRIEFING_TTL_HOURS = 24

# Validator regeneration attempts before falling back to templated
# briefing.
MAX_REGENERATION_ATTEMPTS = 2

# CloudWatch namespace for treatment-response metrics. Slice by
# treatment_pair_id, cohort axis, OOD severity, and disagreement
# flag to catch subgroup drift.
METRIC_NAMESPACE = "TreatmentResponseRecommender"

# Model risk tier thresholds. The catalog tags each pair with a
# tier; tier-1 may qualify for the 21st Century Cures Act CDS
# exemption, tier-2 and above are likely SaMD with corresponding
# regulatory documentation requirements.
MODEL_RISK_TIERS = {
    "tier_1_advisory_only_well_evidenced": "low_risk",
    "tier_2_advisory_observational":       "moderate_risk",
    "tier_3_higher_stakes":                "high_risk",
}

# Synthetic treatment catalog. Three drug classes covering the
# T2D second-line therapy decision space from the recipe's opening
# narrative. Production catalog has dozens of treatments across
# multiple conditions; the example keeps it small for tractability.
SAMPLE_TREATMENT_CATALOG = [
    {
        "treatment_id":     "glp1_receptor_agonist_class",
        "display_name":     "GLP-1 Receptor Agonist (class)",
        "rxnorm_class":     "GLP-1RA",
        "primary_indication": "type_2_diabetes",
        "formulary_status": "tier_2_with_prior_auth",
        "supply_status":    "constrained_intermittent",
    },
    {
        "treatment_id":     "sglt2_inhibitor_class",
        "display_name":     "SGLT2 Inhibitor (class)",
        "rxnorm_class":     "SGLT2",
        "primary_indication": "type_2_diabetes",
        "formulary_status": "tier_1",
        "supply_status":    "available",
    },
    {
        "treatment_id":     "sulfonylurea_class",
        "display_name":     "Sulfonylurea (class)",
        "rxnorm_class":     "SU",
        "primary_indication": "type_2_diabetes",
        "formulary_status": "tier_1",
        "supply_status":    "available",
    },
]

# Synthetic treatment-comparison-pair catalog. Each pair specifies
# the target trial protocol, the model-risk tier, the production
# CATE-estimator endpoint pointers, and the governance metadata.
SAMPLE_PAIR_CATALOG = [
    {
        "pair_id":              "t2d-glp1-vs-sglt2",
        "treatment_id":         "glp1_receptor_agonist_class",
        "comparator_id":        "sglt2_inhibitor_class",
        "indication":           "type_2_diabetes_inadequately_controlled_on_metformin",
        "primary_outcome_id":   "a1c_change_at_90_days",
        "secondary_outcomes":   ["weight_change_at_90_days",
                                  "hypoglycemia_event_at_90_days",
                                  "gi_intolerance_discontinuation_60_days"],
        "model_risk_tier":      "tier_2_advisory_observational",
        "evidence_level":       "RCT_supported_with_observational_extension",
        "guideline_references": ["ada_standards_of_care_2026_section_9.4",
                                   "kdigo_2024_diabetes_guidelines"],
        "is_production":        True,
        "production_calibration_status": "production_calibrated_2026Q1",
        "fairness_axes":        ["language", "race_ethnicity_self_report",
                                   "sdoh_cohort", "age_band"],
        "protocol_version":     "2026-v1",
        "feature_set_version":  "v3",
        "production_pair_endpoints": {
            "causal_forest": "trx-cf-t2d-glp1-vs-sglt2-prod",
            "dr_learner":    "trx-dr-t2d-glp1-vs-sglt2-prod",
            "bart":          "trx-bart-t2d-glp1-vs-sglt2-prod",
        },
    },
    {
        "pair_id":              "t2d-glp1-vs-sulfonylurea",
        "treatment_id":         "glp1_receptor_agonist_class",
        "comparator_id":        "sulfonylurea_class",
        "indication":           "type_2_diabetes_inadequately_controlled_on_metformin",
        "primary_outcome_id":   "a1c_change_at_90_days",
        "secondary_outcomes":   ["weight_change_at_90_days",
                                  "hypoglycemia_event_at_90_days"],
        "model_risk_tier":      "tier_2_advisory_observational",
        "evidence_level":       "RCT_supported_with_observational_extension",
        "guideline_references": ["ada_standards_of_care_2026_section_9.4"],
        "is_production":        True,
        "production_calibration_status": "production_calibrated_2026Q1",
        "fairness_axes":        ["language", "race_ethnicity_self_report",
                                   "sdoh_cohort", "age_band"],
        "protocol_version":     "2026-v1",
        "feature_set_version":  "v3",
        "production_pair_endpoints": {
            "causal_forest": "trx-cf-t2d-glp1-vs-su-prod",
            "dr_learner":    "trx-dr-t2d-glp1-vs-su-prod",
            "bart":          "trx-bart-t2d-glp1-vs-su-prod",
        },
    },
    {
        "pair_id":              "t2d-sglt2-vs-sulfonylurea",
        "treatment_id":         "sglt2_inhibitor_class",
        "comparator_id":        "sulfonylurea_class",
        "indication":           "type_2_diabetes_inadequately_controlled_on_metformin",
        "primary_outcome_id":   "a1c_change_at_90_days",
        "secondary_outcomes":   ["weight_change_at_90_days",
                                  "hypoglycemia_event_at_90_days",
                                  "egfr_change_at_180_days"],
        "model_risk_tier":      "tier_2_advisory_observational",
        "evidence_level":       "RCT_supported_with_observational_extension",
        "guideline_references": ["ada_standards_of_care_2026_section_9.4",
                                   "kdigo_2024_diabetes_guidelines"],
        "is_production":        True,
        "production_calibration_status": "production_calibrated_2026Q1",
        "fairness_axes":        ["language", "race_ethnicity_self_report",
                                   "sdoh_cohort", "age_band"],
        "protocol_version":     "2026-v1",
        "feature_set_version":  "v3",
        "production_pair_endpoints": {
            "causal_forest": "trx-cf-t2d-sglt2-vs-su-prod",
            "dr_learner":    "trx-dr-t2d-sglt2-vs-su-prod",
            "bart":          "trx-bart-t2d-sglt2-vs-su-prod",
        },
    },
]

# Target trial protocol per pair. Specifies eligibility, washout,
# exposure assignment, censoring, and outcome timing. Production
# stores this in the treatment-comparison-pairs table; the example
# inlines it for clarity.
SAMPLE_PROTOCOLS = {
    "t2d-glp1-vs-sglt2": {
        "version":       "2026-v1",
        "eligibility": {
            "diagnosis_required":            "type_2_diabetes",
            "current_medication_required":   "metformin_monotherapy",
            "a1c_min_at_index":              7.0,
            "a1c_max_at_index":              12.0,
            "egfr_min_at_index":             30,
            "exclusions":                    ["type_1_diabetes", "pregnancy",
                                               "active_malignancy",
                                               "dialysis_dependent"],
            "continuous_enrollment_months":  6,
        },
        "washout": {
            "days":              180,
            "excluded_exposures": ["glp1_receptor_agonist_class",
                                    "sglt2_inhibitor_class",
                                    "dpp4_inhibitor_class",
                                    "thiazolidinedione_class"],
        },
        "exposure_definitions": {
            "treated":    "first_dispense_glp1_within_30_days_of_index",
            "comparator": "first_dispense_sglt2_within_30_days_of_index",
        },
        "outcomes": [
            {"outcome_id":   "a1c_change_at_90_days",
              "measurement_window_days": 90,
              "tolerance_days": 14,
              "censor_on_treatment_switch": True,
              "censor_on_discontinuation": False},
            {"outcome_id":   "weight_change_at_90_days",
              "measurement_window_days": 90,
              "tolerance_days": 14},
            {"outcome_id":   "hypoglycemia_event_at_90_days",
              "measurement_window_days": 90,
              "binary": True},
            {"outcome_id":   "gi_intolerance_discontinuation_60_days",
              "measurement_window_days": 60,
              "binary": True},
        ],
        "feature_set_version":           "v3",
        "propensity_hyperparameters":    {"max_depth": 6, "n_estimators": 400,
                                            "learning_rate": 0.05},
        "outcome_hyperparameters": {
            "a1c_change_at_90_days":     {"max_depth": 6, "n_estimators": 600,
                                            "learning_rate": 0.05},
        },
        "cate_hyperparameters": {
            "causal_forest": {"n_estimators": 1000, "min_samples_leaf": 20,
                                "honest": True},
            "dr_learner":    {"max_depth": 5, "n_estimators": 500},
            "bart":          {"n_trees": 200, "n_chains": 4, "n_samples": 2000},
        },
    },
}

# Sensitivity-analysis bounds per pair. Pre-computed at training
# time and used at scoring time to widen reported CIs. The example
# uses a fixed multiplier; production runs E-value computations and
# Rosenbaum bounds per pair.
SAMPLE_SENSITIVITY_BOUNDS = {
    "t2d-glp1-vs-sglt2":     {"e_value": 1.84, "ci_widen_multiplier": 1.20},
    "t2d-glp1-vs-sulfonylurea": {"e_value": 1.21, "ci_widen_multiplier": 1.45},
    "t2d-sglt2-vs-sulfonylurea": {"e_value": 1.62, "ci_widen_multiplier": 1.25},
}

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 scoring pipeline. 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.
    """
    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, bool):
            out[k] = v
        elif isinstance(v, (int, float)):
            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_scoring_run_id(patient_id: str, run_date: str) -> str:
    """
    Scoring run ID used to join scoring results to briefings to
    decisions to outcomes.

    NOTE: This example uses a readable string for clarity. Production
    must replace this with an opaque, non-reversible identifier (UUID
    or HMAC-SHA256 over the composite with a per-environment secret).
    Plain-text patient_ids embedded in scoring IDs (carried in
    EHR responses, scoring API responses, briefings, and decision
    events) are PHI leakage. Mirror the language from 4.4 through 4.7.
    """
    return f"score-{run_date}-{patient_id}-{uuid.uuid4().hex[:8]}"

def _make_briefing_id() -> str:
    """Opaque briefing identifier."""
    return f"brief-{uuid.uuid4().hex[:16]}"

def _make_decision_id() -> str:
    """Opaque decision identifier."""
    return f"decision-{uuid.uuid4().hex[:16]}"

def _make_cohort_id(pair_id: str, protocol_version: str, run_date: str) -> str:
    """Cohort identifier joining cohort archive to model artifacts."""
    return f"cohort-{pair_id}-{protocol_version}-{run_date}"

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")
        time.sleep(2)

def _redact_identifiers(items: list) -> list:
    """
    Strip patient/clinician identifiers from a list of records before
    sending to an LLM. The LLM doesn't need them, and stripping at the
    boundary limits any vendor-side logging exposure (Bedrock service
    terms commit to not training on prompts, but defense-in-depth
    still applies).
    """
    redacted = []
    for item in items:
        copy = dict(item)
        for field in ("patient_id", "clinician_id", "scoring_run_id",
                       "briefing_id", "decision_id"):
            copy.pop(field, None)
        redacted.append(copy)
    return redacted

def _cohort_features_from_profile(patient: dict) -> dict:
    """Pull cohort features for fairness instrumentation from the profile."""
    return {
        "language":                 patient.get("preferred_language", "en"),
        "race_ethnicity_self_report": patient.get(
            "race_ethnicity_self_report", "unknown"),
        "sdoh_cohort":              patient.get("sdoh_cohort", "unknown"),
        "age_band":                 patient.get("age_band", "unknown"),
    }

def _lookup_pair(pair_id: str, pair_catalog: list) -> dict:
    """Return the pair spec for a given pair_id, or {} if not found."""
    for pair in pair_catalog:
        if pair["pair_id"] == pair_id:
            return pair
    return {}

def _lookup_treatment(treatment_id: str, treatment_catalog: list) -> dict:
    """Return the treatment spec for a given treatment_id, or {}."""
    for tx in treatment_catalog:
        if tx["treatment_id"] == treatment_id:
            return tx
    return {}

def construct_cohort(pair: dict, run_date: str,
                       protocols: dict = None) -> dict:
    """
    Run target trial emulation for a single treatment-comparator pair.

    Steps:
      A. Identify candidate patients from the data lake using the
         protocol's eligibility predicates.
      B. Apply the washout window to exclude patients with relevant
         prior exposures.
      C. Assign treatment exposure to each eligible patient based on
         what they were actually prescribed at the index date.
      D. Construct outcome labels at the protocol-specified timing,
         with censoring rules.
      E. Persist the cohort to S3 (versioned, partitioned by pair
         and protocol version).
      F. Persist cohort metadata for traceability.

    Returns the cohort metadata record (without the full cohort,
    which is in S3).
    """
    pair_id = pair["pair_id"]
    protocols = protocols or SAMPLE_PROTOCOLS
    protocol = protocols.get(pair_id, {})
    if not protocol:
        logger.warning("No protocol for pair %s; skipping", pair_id)
        return {}

    protocol_version = protocol["version"]
    cohort_id = _make_cohort_id(pair_id, protocol_version, run_date)

    # Step 1A: candidate patients. Production: parameterized SQL on
    # the data lake joining the eligibility view with the encounter,
    # diagnosis, medication, lab, and enrollment tables. The example
    # delegates to a stub that returns a synthetic cohort.
    candidates = _athena_candidate_query(protocol, run_date)

    # Step 1B: washout. Exclude patients with relevant prior
    # exposures within the washout window. Per-pair washout because
    # what counts as a relevant prior exposure depends on what is
    # being compared.
    eligible = _athena_washout_query(candidates, protocol, run_date)

    # Step 1C: assign exposure. For each eligible patient, determine
    # whether they fell into the treated or comparator arm based on
    # what was actually prescribed at the index date. Patients who
    # received neither (or both within the protocol's grace window)
    # are excluded from this pair's cohort.
    cohort_treated = _assign_arm(
        eligible, protocol["exposure_definitions"]["treated"], "treated",
    )
    cohort_comparator = _assign_arm(
        eligible, protocol["exposure_definitions"]["comparator"], "comparator",
    )
    cohort = cohort_treated + cohort_comparator

    # Step 1D: construct outcome labels. For each cohort member,
    # compute outcomes at the protocol-specified timing with
    # censoring rules.
    cohort_with_outcomes = []
    for member in cohort:
        outcomes = {}
        for outcome_def in protocol["outcomes"]:
            outcomes[outcome_def["outcome_id"]] = _compute_outcome(
                member, outcome_def, protocol,
            )
        cohort_with_outcomes.append({
            "patient_id":      member["patient_id"],
            "treatment_arm":    member["treatment_arm"],
            "index_date":       member["index_date"],
            "covariates":       member["covariates"],   # feature vector at index
            "outcomes":         outcomes,
            "protocol_version": protocol_version,
        })

    # Step 1E: persist cohort to S3 (in production, written as
    # Parquet partitioned by pair_id and protocol_version). The
    # example writes a JSON line list for inspection.
    cohort_s3_path = (
        f"s3://{TRX_COHORTS_BUCKET}/pair={pair_id}/"
        f"protocol_version={protocol_version}/run_date={run_date}/cohort.jsonl"
    )
    try:
        s3_client.put_object(
            Bucket=TRX_COHORTS_BUCKET,
            Key=(f"pair={pair_id}/protocol_version={protocol_version}/"
                  f"run_date={run_date}/cohort.jsonl"),
            Body="\n".join(
                json.dumps(rec, default=str) for rec in cohort_with_outcomes
            ).encode("utf-8"),
        )
    except Exception as exc:
        logger.warning(
            "Failed to persist cohort for %s: %s; continuing with metadata only",
            pair_id, exc,
        )

    treated_size    = sum(1 for r in cohort_with_outcomes
                            if r["treatment_arm"] == "treated")
    comparator_size = sum(1 for r in cohort_with_outcomes
                            if r["treatment_arm"] == "comparator")

    metadata = {
        "cohort_id":         cohort_id,
        "pair_id":           pair_id,
        "protocol_version":  protocol_version,
        "run_date":          run_date,
        "cohort_path":       cohort_s3_path,
        "size_treated":      treated_size,
        "size_comparator":   comparator_size,
        "outcome_definitions": [o["outcome_id"] for o in protocol["outcomes"]],
        "feature_set_version": protocol["feature_set_version"],
        "constructed_at":    _now_iso(),
    }

    # Step 1F: persist cohort metadata for traceability.
    metadata_table = dynamodb.Table(COHORT_METADATA_TABLE)
    try:
        metadata_table.put_item(Item=_to_decimal_dict(metadata))
    except Exception as exc:
        logger.warning(
            "Failed to persist cohort metadata for %s: %s", cohort_id, exc,
        )

    try:
        kinesis_client.put_record(
            StreamName=TRX_EVENTS_STREAM_NAME,
            PartitionKey=pair_id,
            Data=json.dumps({
                "event_type":      "cohort_constructed",
                "cohort_id":       cohort_id,
                "pair_id":         pair_id,
                "size_treated":    treated_size,
                "size_comparator": comparator_size,
                "timestamp":       _now_iso(),
            }, default=str).encode("utf-8"),
        )
    except Exception as exc:
        logger.warning(
            "Failed to publish cohort_constructed event: %s", exc,
        )

    logger.info(
        "Cohort %s: %d treated, %d comparator",
        cohort_id, treated_size, comparator_size,
    )
    return metadata

def _athena_candidate_query(protocol: dict, run_date: str) -> list:
    """
    Production: Athena query on the data lake using the protocol's
    eligibility predicates. Demo: return the synthetic cohort.
    """
    return _DEMO_CANDIDATE_COHORT

def _athena_washout_query(candidates: list, protocol: dict,
                            run_date: str) -> list:
    """
    Production: Athena query joining candidate patients with the
    medication-exposure view to apply the washout. Demo: stub
    that returns candidates as-is, since the synthetic cohort is
    pre-filtered.
    """
    return candidates

def _assign_arm(eligible: list, exposure_definition: str, arm: str) -> list:
    """
    Apply the exposure definition to assign each eligible patient
    to the treated or comparator arm. Production: SQL pattern that
    looks at the medication-dispense view; the example uses a
    pre-tagged synthetic cohort.
    """
    out = []
    for patient in eligible:
        if patient.get("synthetic_treatment_arm") == arm:
            out.append({
                **patient,
                "treatment_arm": arm,
            })
    return out

def _compute_outcome(member: dict, outcome_def: dict,
                      protocol: dict) -> dict:
    """
    Compute the outcome value for a cohort member at the
    protocol-specified timing, with censoring. Production: pull
    longitudinal lab, encounter, and medication data; apply the
    outcome definition's measurement window and tolerance; flag
    censoring on treatment switch, plan disenrollment, or death.
    The example reads from a synthetic per-patient outcome dict.
    """
    outcome_id = outcome_def["outcome_id"]
    raw = member.get("synthetic_outcomes", {}).get(outcome_id)
    if raw is None:
        return {
            "value":         None,
            "observed":      False,
            "censored":      True,
            "censor_reason": "no_observed_outcome_in_window",
        }
    return {
        "value":         raw["value"],
        "observed":      raw.get("observed", True),
        "censored":      raw.get("censored", False),
        "censor_reason": raw.get("censor_reason"),
    }

# Synthetic candidate cohort populated by the runner at the bottom
# of this file. In production this comes from Athena.
_DEMO_CANDIDATE_COHORT: list = []

def train_pair_models(cohort_metadata: dict, run_date: str) -> dict:
    """
    Train the propensity, outcome, and CATE-ensemble models for a
    single treatment-comparator pair.

    Production: each training stage launches a SageMaker Training
    Job with a BYOC container that wraps EconML, grf, or bartCause.
    The example simulates training by registering pseudo-model ARNs
    and computing a synthetic propensity-overlap diagnostic.

    Returns a dict with the training status and the registered model
    pointers, or {"status": "suspended", "reason": "..."} if the
    propensity-overlap check fails.
    """
    cohort_id = cohort_metadata["cohort_id"]
    pair_id   = cohort_metadata["pair_id"]
    pair      = _lookup_pair(pair_id, SAMPLE_PAIR_CATALOG)

    # ---- Stage A: propensity score model ----
    # Predicts P(treatment = treated | covariates X). Trained on
    # the full cohort with treatment_arm as the binary label.
    # XGBoost with isotonic calibration is a reasonable baseline;
    # logistic regression is the simplest comparator. Production
    # cross-validates and evaluates calibration on a held-out set.
    propensity_model_arn = _simulate_training_job(
        algorithm="xgboost-with-isotonic-calibration",
        cohort_id=cohort_id,
        target="treatment_arm_treated",
        family="propensity",
        run_date=run_date,
        pair_id=pair_id,
    )

    # Propensity overlap diagnostic. If the treated and comparator
    # cohorts have substantially non-overlapping covariate
    # distributions, the CATE estimator is extrapolating across the
    # gap, and the resulting estimates are dominated by modeling
    # assumptions rather than data. This is a HARD GATE: insufficient
    # overlap suspends training for this pair until the cohort can
    # be re-scoped (different eligibility, different comparator).
    overlap = _assess_propensity_overlap(propensity_model_arn,
                                            cohort_metadata)
    if overlap["severe"]:
        logger.warning(
            "Propensity overlap severe for %s; suspending training",
            pair_id,
        )
        try:
            kinesis_client.put_record(
                StreamName=TRX_EVENTS_STREAM_NAME,
                PartitionKey=pair_id,
                Data=json.dumps({
                    "event_type": "training_suspended",
                    "cohort_id":  cohort_id,
                    "pair_id":    pair_id,
                    "reason":     "propensity_overlap_severe",
                    "details":    overlap,
                    "timestamp":  _now_iso(),
                }, default=str).encode("utf-8"),
            )
        except Exception:
            pass
        return {
            "status":  "suspended",
            "reason":  "propensity_overlap_severe",
            "details": overlap,
        }

    # ---- Stage B: outcome model(s) ----
    # Predicts E[Y | X, T] for each outcome of interest. The
    # outcome model is one of the inputs to the meta-learners
    # (DR-learner, X-learner). Trained on the full cohort with
    # outcome as the label and treatment as a feature.
    outcome_model_arns = {}
    for outcome_id in cohort_metadata["outcome_definitions"]:
        outcome_model_arns[outcome_id] = _simulate_training_job(
            algorithm="xgboost-regression",
            cohort_id=cohort_id,
            target=outcome_id,
            family="outcome",
            run_date=run_date,
            pair_id=pair_id,
        )

    # ---- Stage C: CATE estimator ensemble ----
    # At least two methods from different families. The ensemble
    # surfaces estimator disagreement, which is a structural-
    # uncertainty signal that no single estimator can provide.
    cate_estimator_arns = {}
    for estimator_name in CATE_ESTIMATOR_NAMES:
        cate_estimator_arns[estimator_name] = _simulate_training_job(
            algorithm=f"cate-{estimator_name}",
            cohort_id=cohort_id,
            target="treatment_effect",
            family=f"cate_{estimator_name}",
            run_date=run_date,
            pair_id=pair_id,
        )

    # Persist training status to the pairs table for the governance
    # gate to read in Step 3.
    pairs_table = dynamodb.Table(TREATMENT_COMPARISON_PAIRS_TABLE)
    try:
        pairs_table.update_item(
            Key={"pair_id": pair_id},
            UpdateExpression=(
                "SET training_status = :st, "
                "propensity_model_arn = :pm, "
                "outcome_model_arns = :om, "
                "cate_estimator_arns = :cm, "
                "cohort_id = :cid, "
                "last_training_run = :run"
            ),
            ExpressionAttributeValues=_to_decimal_dict({
                ":st":  "trained_pending_evaluation",
                ":pm":  propensity_model_arn,
                ":om":  outcome_model_arns,
                ":cm":  cate_estimator_arns,
                ":cid": cohort_id,
                ":run": run_date,
            }),
        )
    except Exception as exc:
        logger.warning(
            "Failed to update training status for %s: %s", pair_id, exc,
        )

    return {
        "status":               "trained_pending_evaluation",
        "pair_id":              pair_id,
        "cohort_id":            cohort_id,
        "propensity_model_arn": propensity_model_arn,
        "outcome_model_arns":   outcome_model_arns,
        "cate_estimator_arns":  cate_estimator_arns,
        "propensity_overlap":   overlap,
    }

def _simulate_training_job(algorithm: str, cohort_id: str, target: str,
                              family: str, run_date: str,
                              pair_id: str) -> str:
    """
    Stand-in for SageMaker Training Job submission. Production:
    `sagemaker_client.create_training_job(...)` with a BYOC
    container, then poll until the job completes, then register
    the resulting model in the SageMaker Model Registry.
    """
    job_name = f"{family}-{pair_id}-{run_date}"
    pseudo_arn = (
        f"arn:aws:sagemaker:us-east-1:000000000000:model/{job_name}"
    )
    logger.info(
        "Simulated training: algorithm=%s job=%s target=%s",
        algorithm, job_name, target,
    )
    return pseudo_arn

def _assess_propensity_overlap(propensity_model_arn: str,
                                  cohort_metadata: dict) -> dict:
    """
    Compute a propensity-overlap diagnostic for the cohort.

    Production: predict propensity scores for every cohort member,
    inspect the distributions in the treated and comparator arms,
    and flag severe overlap failure when the proportion of patients
    with propensity in the tails (< 0.05 or > 0.95) exceeds a
    pre-set threshold (often 5%-10%, depending on cohort size).

    The example computes a stub that uses the cohort sizes as a
    proxy: very imbalanced cohorts get worse overlap scores. Real
    propensity overlap is a much richer diagnostic.
    """
    treated   = cohort_metadata["size_treated"]
    comparator = cohort_metadata["size_comparator"]
    total = treated + comparator
    if total == 0:
        return {"severe": True, "imbalance": 1.0,
                 "tail_fraction_estimate": 1.0}
    imbalance = abs(treated - comparator) / total

    # Synthetic tail-fraction estimate. Production: actual percentiles
    # of the predicted propensity distribution.
    tail_fraction = 0.05 + 0.40 * imbalance
    return {
        "severe":               tail_fraction > 0.20,
        "imbalance":            round(imbalance, 4),
        "tail_fraction_estimate": round(tail_fraction, 4),
        "treated_size":         treated,
        "comparator_size":      comparator,
    }

def evaluate_and_gate_pair_models(pair_id: str, run_date: str) -> dict:
    """
    Run calibration, cohort-stratified fairness, estimator-agreement,
    and sensitivity analyses for a single pair. Persist the
    evaluation report and create a governance-review-task for human
    review before promotion.

    Returns the evaluation report plus the review task id.
    """
    pairs_table = dynamodb.Table(TREATMENT_COMPARISON_PAIRS_TABLE)
    pair = _from_decimal(pairs_table.get_item(
        Key={"pair_id": pair_id}
    ).get("Item") or {})
    if not pair:
        logger.warning("Pair %s not found in registry", pair_id)
        return {}

    # ---- Stage A: held-out calibration ----
    # Production: use the held-out test split of the training cohort
    # (typically temporal: train on older cohort, test on newer).
    # Predict treatment effects on the test set and compare against
    # observed effects within covariate-defined subgroups.
    calibration_results = {}
    for estimator_name in CATE_ESTIMATOR_NAMES:
        calibration_results[estimator_name] = _compute_calibration(
            estimator_name, pair, run_date,
        )

    # ---- Stage B: cohort-stratified fairness ----
    # Repeat the calibration within each protected cohort. Calibration
    # parity across cohorts is the bar.
    fairness_results = {}
    for axis in pair.get("fairness_axes", []):
        for cohort_value in _DEMO_COHORT_VALUES.get(axis, []):
            sample_size = _DEMO_FAIRNESS_SAMPLE_SIZE.get(
                (pair_id, axis, cohort_value), 0,
            )
            if sample_size < MIN_FAIRNESS_SAMPLE_SIZE:
                fairness_results[(axis, cohort_value)] = {
                    "status":      "insufficient_sample",
                    "sample_size": sample_size,
                }
                continue
            fairness_results[(axis, cohort_value)] = _compute_calibration_subset(
                pair_id, axis, cohort_value, sample_size,
            )

    # ---- Stage C: estimator agreement ----
    # Per-patient correlation of CATE estimates across the ensemble.
    # Disagreement is a structural-uncertainty signal.
    agreement = _compute_estimator_agreement(pair_id)

    # ---- Stage D: sensitivity analysis ----
    # Pre-compute E-value and ci-widening multiplier per pair.
    # Production runs VanderWeele-Ding E-value computation and
    # Rosenbaum bounds on the actual cohort.
    sensitivity = SAMPLE_SENSITIVITY_BOUNDS.get(pair_id, {
        "e_value": 1.20,
        "ci_widen_multiplier": 1.30,
    })

    # ---- Stage E: classify overall ----
    overall_status = _classify_overall_evaluation(
        calibration_results, fairness_results, agreement, sensitivity,
    )

    eval_report = {
        "pair_id":            pair_id,
        "run_date":           run_date,
        "calibration_results": calibration_results,
        "fairness_results":    {f"{k[0]}={k[1]}": v
                                  for k, v in fairness_results.items()},
        "agreement_results":   agreement,
        "sensitivity_results": sensitivity,
        "evaluation_status":   overall_status,
    }

    # Persist the evaluation report.
    eval_path = (
        f"s3://{TRX_EVAL_BUCKET}/pair={pair_id}/run={run_date}/report.json"
    )
    try:
        s3_client.put_object(
            Bucket=TRX_EVAL_BUCKET,
            Key=f"pair={pair_id}/run={run_date}/report.json",
            Body=json.dumps(eval_report, default=str).encode("utf-8"),
        )
    except Exception as exc:
        logger.warning(
            "Failed to persist eval report for %s: %s", pair_id, exc,
        )

    # ---- Stage F: governance gate ----
    # The system creates a review task; the cross-functional
    # committee (medical director, clinical informatics, equity
    # lead, data science lead) decides whether to promote.
    task_id = f"review-{uuid.uuid4().hex[:16]}"
    review_tasks_table = dynamodb.Table(GOVERNANCE_REVIEW_TASKS_TABLE)
    try:
        review_tasks_table.put_item(Item=_to_decimal_dict({
            "task_id":              task_id,
            "pair_id":              pair_id,
            "run_date":             run_date,
            "evaluation_report_path": eval_path,
            "evaluation_status":    overall_status,
            "status":               "pending_review",
            "created_at":           run_date,
        }))
    except Exception as exc:
        logger.warning(
            "Failed to persist governance review task: %s", exc,
        )

    try:
        kinesis_client.put_record(
            StreamName=TRX_EVENTS_STREAM_NAME,
            PartitionKey=pair_id,
            Data=json.dumps({
                "event_type":        "governance_review_pending",
                "pair_id":           pair_id,
                "task_id":           task_id,
                "evaluation_status": overall_status,
                "timestamp":         _now_iso(),
            }, default=str).encode("utf-8"),
        )
    except Exception as exc:
        logger.warning(
            "Failed to publish governance_review_pending: %s", exc,
        )

    return {
        "report":  eval_report,
        "task_id": task_id,
    }

def process_governance_decision(task_id: str, human_decision: dict) -> None:
    """
    Process a committee-reviewed governance decision. On approval,
    promote the new model artifacts to production aliases in the
    Model Registry and update production_pair_endpoints in the
    pairs table. On rejection, mark training as evaluation_failed.

    human_decision is one of:
      { "decision": "approve", "reviewer_ids": [...], "notes": "..." }
      { "decision": "reject", "reviewer_ids": [...],
        "rejection_reason": "..." }
    """
    review_tasks_table = dynamodb.Table(GOVERNANCE_REVIEW_TASKS_TABLE)
    pairs_table = dynamodb.Table(TREATMENT_COMPARISON_PAIRS_TABLE)

    task = _from_decimal(review_tasks_table.get_item(
        Key={"task_id": task_id}
    ).get("Item") or {})
    if not task:
        logger.warning("Governance review task %s not found", task_id)
        return

    pair_id = task["pair_id"]

    if human_decision.get("decision") == "approve":
        # Promote: in production, alias the new model artifacts to
        # production endpoints, update production_pair_endpoints in
        # the pairs table, archive the prior production artifacts
        # for rollback.
        try:
            pairs_table.update_item(
                Key={"pair_id": pair_id},
                UpdateExpression=(
                    "SET training_status = :st, "
                    "is_production = :prod, "
                    "production_calibration_status = :pcs, "
                    "promoted_at = :pa"
                ),
                ExpressionAttributeValues=_to_decimal_dict({
                    ":st":  "production",
                    ":prod": True,
                    ":pcs": f"production_calibrated_{task['run_date']}",
                    ":pa":  _now_iso(),
                }),
            )
        except Exception as exc:
            logger.warning(
                "Failed to promote pair %s: %s", pair_id, exc,
            )

        try:
            kinesis_client.put_record(
                StreamName=TRX_EVENTS_STREAM_NAME,
                PartitionKey=pair_id,
                Data=json.dumps({
                    "event_type": "model_promoted",
                    "pair_id":    pair_id,
                    "task_id":    task_id,
                    "promoted_by": human_decision.get("reviewer_ids", []),
                    "timestamp":  _now_iso(),
                }, default=str).encode("utf-8"),
            )
        except Exception:
            pass

    else:
        try:
            pairs_table.update_item(
                Key={"pair_id": pair_id},
                UpdateExpression="SET training_status = :st",
                ExpressionAttributeValues={
                    ":st": "evaluation_failed",
                },
            )
        except Exception:
            pass

    try:
        review_tasks_table.update_item(
            Key={"task_id": task_id},
            UpdateExpression=(
                "SET #s = :resolved, human_decision = :hd, "
                "resolved_at = :ra"
            ),
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues=_to_decimal_dict({
                ":resolved": "resolved",
                ":hd":       human_decision,
                ":ra":       _now_iso(),
            }),
        )
    except Exception:
        pass

def _compute_calibration(estimator_name: str, pair: dict,
                            run_date: str) -> dict:
    """
    Compute calibration on the held-out test set for a single
    estimator. Production: bin patients by predicted treatment
    effect, compare predicted-vs-observed mean within each bin,
    fit a calibration slope and intercept. Demo: synthetic.
    """
    base = _DEMO_CALIBRATION_BY_ESTIMATOR.get(estimator_name, {
        "calibration_slope":     0.92,
        "calibration_intercept": 0.05,
        "n_test_patients":       300,
    })
    slope = base["calibration_slope"]
    return {
        **base,
        "calibration_warning": slope < CALIBRATION_SLOPE_LOW
                                or slope > CALIBRATION_SLOPE_HIGH,
    }

def _compute_calibration_subset(pair_id: str, axis: str,
                                  cohort_value: str,
                                  sample_size: int) -> dict:
    """
    Compute calibration within a single protected cohort. Production:
    same calibration computation as above, restricted to the cohort
    subset. Demo: synthetic.
    """
    key = (pair_id, axis, cohort_value)
    base = _DEMO_COHORT_CALIBRATION.get(key, {
        "calibration_slope":     0.88,
        "calibration_intercept": 0.04,
    })
    return {
        **base,
        "sample_size":         sample_size,
        "cohort_axis":         axis,
        "cohort_value":        cohort_value,
        "calibration_warning": base["calibration_slope"] < CALIBRATION_SLOPE_LOW
                                or base["calibration_slope"] > CALIBRATION_SLOPE_HIGH,
    }

def _compute_estimator_agreement(pair_id: str) -> dict:
    """
    Per-patient agreement among ensemble members. Production:
    correlate per-patient point estimates from each estimator,
    flag the percent of patients with inter-estimator spread above
    DISAGREEMENT_THRESHOLD. Demo: synthetic.
    """
    return _DEMO_ESTIMATOR_AGREEMENT.get(pair_id, {
        "mean_pairwise_correlation":    0.78,
        "percent_high_disagreement":    0.14,
        "agreement_score_summary":      "moderate",
    })

def _classify_overall_evaluation(calibration_results: dict,
                                    fairness_results: dict,
                                    agreement: dict,
                                    sensitivity: dict) -> str:
    """
    Classify the evaluation report as green (clear pass), yellow
    (warnings, governance review required), or red (clear fail,
    suspend). Production: explicit policy with thresholds reviewed
    by the committee; the demo uses a coarse heuristic.
    """
    cal_warnings = sum(
        1 for r in calibration_results.values()
        if isinstance(r, dict) and r.get("calibration_warning")
    )
    fairness_warnings = sum(
        1 for r in fairness_results.values()
        if isinstance(r, dict) and r.get("calibration_warning")
    )
    if cal_warnings == 0 and fairness_warnings == 0:
        if agreement.get("percent_high_disagreement", 1.0) <= 0.10:
            return "green"
        return "yellow"
    if cal_warnings >= 2 or fairness_warnings >= 2:
        return "red"
    return "yellow"

# Demo state populated by the runner.
_DEMO_COHORT_VALUES: dict = {}
_DEMO_FAIRNESS_SAMPLE_SIZE: dict = {}
_DEMO_CALIBRATION_BY_ESTIMATOR: dict = {}
_DEMO_COHORT_CALIBRATION: dict = {}
_DEMO_ESTIMATOR_AGREEMENT: dict = {}

def score_patient(patient_id: str, request_context: dict,
                    patients: dict, pair_catalog: list,
                    treatment_catalog: list) -> dict:
    """
    Run on-demand scoring for an index patient.

    Steps:
      A. Identify eligible treatment-comparator pairs from the catalog.
      B. For each eligible pair, invoke the CATE ensemble, retrieve
         the similar-patient cohort summary, compute uncertainty,
         and flag OOD.
      C. Apply sensitivity-analysis bounds to widen reported CIs.
      D. Persist scoring result; emit scoring event.

    Returns the structured scoring result for downstream briefing
    generation.
    """
    patient_features = _featurestore_get_patient(patient_id, patients)
    if not patient_features:
        logger.warning("No features for patient %s", patient_id)
        return {}

    # Step 4A: identify eligible pairs.
    eligible_pairs = []
    for pair in pair_catalog:
        if not pair.get("is_production"):
            continue
        if not _meets_pair_eligibility(patient_features, pair, request_context):
            continue
        if _has_contraindication(patient_features, pair):
            continue
        eligible_pairs.append(pair)

    scoring_run_id = _make_scoring_run_id(patient_id, _today_str())

    if not eligible_pairs:
        # No eligible pairs. Return a structured response that tells
        # the clinician we have nothing to add for this patient. Do
        # not silently produce a low-confidence estimate.
        result = {
            "patient_id":     patient_id,
            "scoring_run_id": scoring_run_id,
            "scoring_status": "no_eligible_pairs",
            "scoring_reason": "no_pairs_match_index_condition_and_eligibility",
            "request_context": request_context,
            "scoring_completed_at": _now_iso(),
        }
        _persist_scoring_result(result)
        return result

    # Step 4B: per-pair scoring.
    pair_results = []
    for pair in eligible_pairs:
        # Run each estimator. Real-time endpoints return per-estimator
        # point estimate and confidence interval.
        estimator_outputs = {}
        for estimator_name in CATE_ESTIMATOR_NAMES:
            endpoint = pair["production_pair_endpoints"].get(estimator_name)
            estimator_outputs[estimator_name] = _invoke_cate_endpoint(
                endpoint, patient_features, pair["pair_id"],
            )

        # Step 4C: ensemble uncertainty.
        ensemble = _combine_ensemble_estimates(estimator_outputs)

        # Step 4D: similar-patient cohort retrieval (summary only;
        # full cohort would be PHI-leaking).
        cohort_summary = _retrieve_similar_patient_summary(
            patient_features, pair["pair_id"],
        )

        # Step 4E: out-of-distribution flag.
        ood_flag = _compute_ood_flag(
            patient_features, pair["pair_id"], cohort_summary,
        )

        # Step 4F: apply sensitivity-analysis bounds.
        adjusted = _apply_sensitivity_bounds(ensemble, pair["pair_id"])

        # If OOD severity is above the suppress threshold, mark
        # this pair's result as suppressed; the briefing layer will
        # explicitly tell the clinician we have no reliable estimate.
        if ood_flag["severity"] >= OOD_SEVERITY_SUPPRESS_THRESHOLD:
            pair_result = {
                "treatment_pair_id":   pair["pair_id"],
                "treatment_id":        pair["treatment_id"],
                "comparator_id":       pair["comparator_id"],
                "outcome_id":          pair["primary_outcome_id"],
                "scoring_status":      "suppressed_oodflag",
                "ood_flag":            ood_flag,
                "evidence_level":      pair["evidence_level"],
                "formulary_status":    _lookup_treatment(
                    pair["treatment_id"], treatment_catalog,
                ).get("formulary_status"),
            }
        else:
            pair_result = {
                "treatment_pair_id":     pair["pair_id"],
                "treatment_id":          pair["treatment_id"],
                "comparator_id":         pair["comparator_id"],
                "outcome_id":            pair["primary_outcome_id"],
                "point_estimate":        round(adjusted["point_estimate"], 4),
                "ci_low":                round(adjusted["ci_low"], 4),
                "ci_high":               round(adjusted["ci_high"], 4),
                "estimator_agreement":   round(
                    ensemble["estimator_agreement_score"], 4),
                "disagreement_flag":     ensemble["disagreement_flag"],
                "cohort_summary":        cohort_summary,
                "ood_flag":              ood_flag,
                "evidence_level":        pair["evidence_level"],
                "formulary_status":      _lookup_treatment(
                    pair["treatment_id"], treatment_catalog,
                ).get("formulary_status"),
                "guideline_references":  pair.get("guideline_references", []),
                "calibration_status":    pair.get(
                    "production_calibration_status"),
            }
        pair_results.append(pair_result)

        # Cohort-sliced metric for the equity dashboard.
        cohort_features = _cohort_features_from_profile(
            patients.get(patient_id, {}))
        _emit_metric(
            "treatment_scoring_completed", value=1,
            dimensions={
                "pair_id":     pair["pair_id"],
                "ood_severity_band": _severity_band(ood_flag["severity"]),
                "language":    cohort_features.get("language", "unknown"),
                "sdoh_cohort": cohort_features.get("sdoh_cohort", "unknown"),
            },
        )

    # Step 4G: persist scoring result.
    result = {
        "patient_id":          patient_id,
        "scoring_run_id":      scoring_run_id,
        "request_context":     request_context,
        "index_condition":     request_context.get("index_condition"),
        "eligible_pair_ids":   [p["pair_id"] for p in eligible_pairs],
        "pair_results":        pair_results,
        "scoring_status":      "completed",
        "scoring_completed_at": _now_iso(),
    }
    _persist_scoring_result(result)

    try:
        kinesis_client.put_record(
            StreamName=TRX_EVENTS_STREAM_NAME,
            PartitionKey=patient_id,
            Data=json.dumps({
                "event_type":         "treatment_scoring_completed",
                "patient_id":         patient_id,
                "scoring_run_id":     scoring_run_id,
                "eligible_pair_count": len(eligible_pairs),
                "any_oodflag":        any(
                    p.get("ood_flag", {}).get("severity", 0)
                      >= OOD_SEVERITY_WARNING_THRESHOLD
                    for p in pair_results),
                "any_disagreement":   any(
                    p.get("disagreement_flag", False) for p in pair_results),
                "timestamp":          _now_iso(),
            }, default=str).encode("utf-8"),
        )
    except Exception as exc:
        logger.warning(
            "Failed to publish treatment_scoring_completed event: %s", exc,
        )

    return result

def _featurestore_get_patient(patient_id: str, patients: dict) -> dict:
    """
    Production: SageMaker Feature Store online-store GetRecord call
    for the patient. Demo: read from the synthetic patients dict.
    """
    return patients.get(patient_id, {})

def _meets_pair_eligibility(patient_features: dict, pair: dict,
                              request_context: dict) -> bool:
    """
    Determine whether the patient meets the pair's eligibility
    predicates. Production: encode the protocol's eligibility logic;
    the demo applies a coarse condition match.
    """
    if request_context.get("index_condition") != pair["indication"]:
        return False
    # Demo: T2D inadequately controlled on metformin.
    if pair["indication"].startswith("type_2_diabetes"):
        if "type_2_diabetes" not in patient_features.get(
            "active_conditions", []):
            return False
        if "metformin" not in patient_features.get("current_medications", []):
            return False
        a1c = patient_features.get("recent_lab_trends", {}).get("a1c_recent")
        if a1c is None or a1c < 7.0 or a1c > 12.0:
            return False
    return True

def _has_contraindication(patient_features: dict, pair: dict) -> bool:
    """
    Demo contraindication check. Production: a richer rules engine
    plus integration with the EHR's medication-allergy and
    drug-interaction services.
    """
    contras = patient_features.get("contraindications", [])
    if pair["treatment_id"] == "glp1_receptor_agonist_class":
        if "history_of_pancreatitis" in contras:
            return True
        if "medullary_thyroid_carcinoma_personal_or_family" in contras:
            return True
    if pair["treatment_id"] == "sglt2_inhibitor_class":
        if "egfr_below_30" in contras:
            return True
        if "active_diabetic_ketoacidosis_history" in contras:
            return True
    return False

def _invoke_cate_endpoint(endpoint: str, patient_features: dict,
                            pair_id: str) -> dict:
    """
    Production: SageMaker Real-Time Inference InvokeEndpoint with
    the patient feature vector serialized for the model. Returns
    point estimate and confidence interval.

    The example uses a rule-based proxy that returns a synthetic
    estimate keyed off the patient's a1c, weight, and pair id.
    """
    a1c = patient_features.get("recent_lab_trends", {}).get(
        "a1c_recent", 8.0)
    bmi = patient_features.get("bmi", 30)

    # Synthetic CATE estimates roughly modeled on T2D second-line
    # therapy literature. Real estimates come from the trained
    # CATE model.
    if pair_id == "t2d-glp1-vs-sglt2":
        # GLP-1 advantage on A1c reduction at 90 days, more pronounced
        # at higher baseline A1c and BMI.
        point = -0.50 - 0.10 * max(0, a1c - 8.0) - 0.005 * max(0, bmi - 28)
        ci_half = 0.30
    elif pair_id == "t2d-glp1-vs-sulfonylurea":
        point = -0.30 - 0.05 * max(0, a1c - 8.0)
        ci_half = 0.35
    elif pair_id == "t2d-sglt2-vs-sulfonylurea":
        point = 0.10 - 0.03 * max(0, a1c - 8.0)
        ci_half = 0.30
    else:
        point = 0.0
        ci_half = 0.50

    # Inject a small amount of estimator-specific variation so the
    # ensemble has nonzero spread.
    variation = (hash(endpoint) % 100) / 1000 - 0.05
    point = point + variation
    return {
        "point_estimate":  point,
        "ci_low":          point - ci_half,
        "ci_high":         point + ci_half,
    }

def _combine_ensemble_estimates(estimator_outputs: dict) -> dict:
    """
    Combine per-estimator estimates into ensemble uncertainty. The
    ensemble point estimate is the mean across estimators; the
    ensemble CI is the union of estimator CIs (the most conservative
    representation of combined sampling and model uncertainty).
    Disagreement is the spread of point estimates normalized to
    the median CI width.
    """
    points = [o["point_estimate"] for o in estimator_outputs.values()]
    ci_lows = [o["ci_low"] for o in estimator_outputs.values()]
    ci_highs = [o["ci_high"] for o in estimator_outputs.values()]

    mean_point = sum(points) / len(points)
    union_low  = min(ci_lows)
    union_high = max(ci_highs)
    spread     = max(points) - min(points)
    median_width = sorted([h - l for h, l in zip(ci_highs, ci_lows)])[
        len(points) // 2]
    if median_width <= 0:
        median_width = 1e-6
    normalized_spread = spread / median_width
    agreement_score   = max(0.0, 1.0 - normalized_spread)

    return {
        "point_estimate":             mean_point,
        "ci_low":                     union_low,
        "ci_high":                    union_high,
        "estimator_agreement_score":  agreement_score,
        "disagreement_flag":          normalized_spread > DISAGREEMENT_THRESHOLD,
    }

def _retrieve_similar_patient_summary(patient_features: dict,
                                          pair_id: str) -> dict:
    """
    Retrieve the similar-patient cohort summary underlying the
    estimate. Production: query the cohort index using the patient
    embedding (or a hand-engineered similarity metric) to return
    summary statistics over the K nearest training-cohort patients.
    Full-cohort retrieval would be PHI-leaking; only summaries leave
    the cohort store.

    The example returns a synthetic summary keyed off the pair id.
    """
    base = _DEMO_COHORT_SUMMARIES.get(pair_id, {})
    if not base:
        return {
            "total_size":              0,
            "cohort_match_quality":    "no_data",
        }
    return {
        **base,
        "training_data_recency":  base.get(
            "training_data_recency", "2023-01-01_to_2025-12-31"),
    }

def _compute_ood_flag(patient_features: dict, pair_id: str,
                        cohort_summary: dict) -> dict:
    """
    Compute the out-of-distribution severity for this patient on
    this pair. Production: combine propensity-tail flag, embedding-
    distance from training distribution, and per-feature density
    estimates.

    Demo: severity rises if cohort match quality is poor or if the
    patient's a1c is outside the typical trained range.
    """
    a1c = patient_features.get("recent_lab_trends", {}).get(
        "a1c_recent", 8.0)
    cohort_size = cohort_summary.get("total_size", 0)
    match_score = cohort_summary.get("demographic_match_score", 0.0)

    severity = 0.0
    reasons = []
    if cohort_size < MIN_SIMILAR_PATIENT_COHORT:
        severity = max(severity, 0.85)
        reasons.append("similar_patient_cohort_too_small")
    if match_score < 0.5:
        severity = max(severity, 0.70)
        reasons.append("low_demographic_match")
    if a1c < 7.0 or a1c > 11.0:
        severity = max(severity, 0.40)
        reasons.append("a1c_outside_training_range")
    return {
        "is_ood":   severity >= OOD_SEVERITY_WARNING_THRESHOLD,
        "severity": round(severity, 3),
        "reasons":  reasons,
    }

def _apply_sensitivity_bounds(ensemble: dict, pair_id: str) -> dict:
    """
    Apply pre-computed sensitivity-analysis bounds to widen the
    reported CI. The widening accounts for unmeasured-confounding
    structural uncertainty that the model's statistical CI doesn't
    capture.
    """
    sensitivity = SAMPLE_SENSITIVITY_BOUNDS.get(pair_id, {
        "ci_widen_multiplier": 1.30,
    })
    multiplier = sensitivity["ci_widen_multiplier"]
    point = ensemble["point_estimate"]
    half = (ensemble["ci_high"] - ensemble["ci_low"]) / 2
    widened = half * multiplier
    return {
        "point_estimate": point,
        "ci_low":         point - widened,
        "ci_high":        point + widened,
    }

def _persist_scoring_result(result: dict) -> None:
    """Persist the scoring result to DynamoDB for the briefing layer."""
    table = dynamodb.Table(SCORING_RESULTS_TABLE)
    try:
        table.put_item(Item=_to_decimal_dict(result))
    except Exception as exc:
        logger.warning(
            "Failed to persist scoring result %s: %s",
            result.get("scoring_run_id"), exc,
        )

def _severity_band(severity: float) -> str:
    """Coarse OOD severity band for metric dimensions."""
    if severity >= OOD_SEVERITY_SUPPRESS_THRESHOLD:
        return "suppress"
    if severity >= OOD_SEVERITY_WARNING_THRESHOLD:
        return "warn"
    return "normal"

# Demo state populated by the runner.
_DEMO_COHORT_SUMMARIES: dict = {}

def generate_briefing(scoring_run_id: str, patients: dict,
                        treatment_catalog: list) -> dict:
    """
    Generate the clinician-facing comparison briefing for a
    scoring run, with strict validator enforcement and templated
    fallback.

    Returns the briefing dict that was persisted.
    """
    scoring_table = dynamodb.Table(SCORING_RESULTS_TABLE)
    scoring = _from_decimal(scoring_table.get_item(
        Key={"patient_id": _scoring_run_patient(scoring_run_id),
              "scoring_run_id": scoring_run_id}
    ).get("Item") or {})

    if not scoring:
        # In production, the scoring-results table is keyed on the
        # composite (patient_id, scoring_run_id). Demo: scan as a
        # fallback so the runner doesn't need to know the patient.
        scoring = _scan_scoring_result(scoring_run_id)

    if not scoring:
        logger.warning("Scoring result %s not found", scoring_run_id)
        return {}

    if scoring.get("scoring_status") == "no_eligible_pairs":
        # No eligible pairs: return a templated "no recommendation"
        # briefing rather than invoking the LLM. The clinician sees
        # an explicit message rather than a confusing empty paragraph.
        briefing_parsed = _no_eligible_briefing(scoring)
        validator_status = True
    else:
        briefing_context = _build_briefing_context(
            scoring, patients, treatment_catalog,
        )

        # Generate via Bedrock with regeneration loop and templated
        # fallback.
        briefing_parsed = None
        validator_status = False

        for attempt in range(MAX_REGENERATION_ATTEMPTS):
            try:
                briefing_parsed = _bedrock_comparison_briefing(
                    briefing_context, strict_mode=(attempt > 0),
                )
            except Exception as exc:
                logger.warning(
                    "Bedrock briefing attempt %d failed: %s", attempt, exc,
                )
                continue

            validation_result = _validate_briefing(
                briefing_parsed, briefing_context,
            )
            if validation_result["passed"]:
                validator_status = True
                break
            else:
                # Provide validator feedback to the next attempt so
                # the LLM has the opportunity to correct the issues.
                briefing_context["validator_feedback"] = (
                    validation_result["feedback"]
                )

        if not validator_status:
            briefing_parsed = _templated_briefing_fallback(briefing_context)
            briefing_parsed["fallback_reason"] = (
                "validator_repeatedly_failed_or_llm_unavailable"
            )
            try:
                kinesis_client.put_record(
                    StreamName=TRX_EVENTS_STREAM_NAME,
                    PartitionKey=scoring["patient_id"],
                    Data=json.dumps({
                        "event_type":     "briefing_validator_fallback",
                        "scoring_run_id": scoring_run_id,
                        "reason":         "validator_repeatedly_failed",
                        "timestamp":      _now_iso(),
                    }, default=str).encode("utf-8"),
                )
            except Exception:
                pass

    briefing_id = _make_briefing_id()
    briefing_record = {
        "briefing_id":      briefing_id,
        "scoring_run_id":   scoring_run_id,
        "patient_id":       scoring["patient_id"],
        "briefing_text":    briefing_parsed,
        "validator_status": validator_status,
        "generated_at":     _now_iso(),
        "ttl_expires_at":   (datetime.datetime.now(timezone.utc)
                              + timedelta(hours=BRIEFING_TTL_HOURS)).isoformat(),
    }

    briefings_table = dynamodb.Table(BRIEFINGS_TABLE)
    try:
        briefings_table.put_item(Item=_to_decimal_dict(briefing_record))
    except Exception as exc:
        logger.warning(
            "Failed to persist briefing %s: %s", briefing_id, exc,
        )

    try:
        kinesis_client.put_record(
            StreamName=TRX_EVENTS_STREAM_NAME,
            PartitionKey=scoring["patient_id"],
            Data=json.dumps({
                "event_type":     "briefing_generated",
                "patient_id":     scoring["patient_id"],
                "scoring_run_id": scoring_run_id,
                "briefing_id":    briefing_id,
                "validator_status": validator_status,
                "timestamp":      _now_iso(),
            }, default=str).encode("utf-8"),
        )
    except Exception:
        pass

    return briefing_record

def _build_briefing_context(scoring: dict, patients: dict,
                              treatment_catalog: list) -> dict:
    """
    Build the structured briefing context. The LLM consumes a
    de-identified version of this; identifiers stay on the
    persistence side.
    """
    patient = patients.get(scoring["patient_id"], {})
    return {
        "patient_summary":    _summarize_patient_for_briefing(patient),
        "index_condition":    scoring.get("index_condition"),
        "pair_results":       scoring.get("pair_results", []),
        "treatments":         [
            _lookup_treatment(p["treatment_id"], treatment_catalog)
            for p in scoring.get("pair_results", [])
        ],
        "any_oodflag":        any(
            p.get("ood_flag", {}).get("severity", 0)
              >= OOD_SEVERITY_WARNING_THRESHOLD
            for p in scoring.get("pair_results", [])),
        "any_disagreement":   any(
            p.get("disagreement_flag", False)
            for p in scoring.get("pair_results", [])),
        "request_context":    scoring.get("request_context", {}),
    }

def _summarize_patient_for_briefing(patient: dict) -> dict:
    """
    Compact patient summary at the level appropriate for clinical
    decision support. Avoids surfacing exact lab values that risk
    being copied into hallucinated text.
    """
    lab_trends = patient.get("recent_lab_trends", {})
    return {
        "age_band":             patient.get("age_band"),
        "active_conditions":    patient.get("active_conditions", []),
        "recent_lab_summary":   {
            "egfr_band":  _egfr_band(lab_trends.get("egfr_recent")),
            "egfr_trend": ("falling"
                            if lab_trends.get("egfr_change_24mo", 0) <= -10
                            else "stable"),
            "a1c_band":   _a1c_band(lab_trends.get("a1c_recent")),
            "albuminuria": ("present"
                              if lab_trends.get("uacr_recent", 0) >= 30
                              else "absent_or_mild"),
        },
        "bmi_band":             _bmi_band(patient.get("bmi")),
        "calcium_score_band":   _calcium_band(
            patient.get("coronary_calcium_score")),
        "current_medications":  patient.get("current_medications", []),
        "social_context":       patient.get("social_context_summary", ""),
        "preferred_language":   patient.get("preferred_language", "en"),
    }

def _a1c_band(a1c) -> str:
    if a1c is None:
        return "unknown"
    if a1c < 7.0:
        return "well_controlled"
    if a1c < 8.0:
        return "moderately_controlled"
    if a1c < 9.0:
        return "elevated"
    return "very_elevated"

def _egfr_band(egfr) -> str:
    if egfr is None:
        return "unknown"
    if egfr >= 90:
        return "normal"
    if egfr >= 60:
        return "mild_decrease"
    if egfr >= 45:
        return "mild_to_moderate"
    if egfr >= 30:
        return "moderate"
    return "severe"

def _bmi_band(bmi) -> str:
    if bmi is None:
        return "unknown"
    if bmi < 25:
        return "normal_or_underweight"
    if bmi < 30:
        return "overweight"
    if bmi < 35:
        return "obesity_class_1"
    if bmi < 40:
        return "obesity_class_2"
    return "obesity_class_3"

def _calcium_band(score) -> str:
    if score is None:
        return "not_measured"
    if score == 0:
        return "zero"
    if score < 100:
        return "minimal"
    if score < 400:
        return "moderate"
    return "high"

def _bedrock_comparison_briefing(context: dict,
                                    strict_mode: bool = False) -> dict:
    """
    Generate a structured clinician-facing comparison briefing via
    Bedrock. The prompt enforces no-recommendation language, explicit
    uncertainty, and required caveats; the validator catches any
    output that slips through.
    """
    de_id = _redact_identifiers([context])[0]

    strictness_addendum = (
        "\n\nSTRICT MODE: A previous attempt failed validation. The "
        "validator caught language that crossed into recommendation. "
        "Be even more careful: describe the comparison without "
        "selecting a treatment. Use phrases like 'the model estimates' "
        "and 'in this cohort' rather than 'we recommend' or 'the "
        "evidence supports'."
        if strict_mode else ""
    )

    prompt = f"""You generate clinician-facing comparison briefings for
a treatment response prediction system. Your role is to package the
deterministic CATE estimates and the patient context into a brief
the clinician reads in about 30 seconds at the point of care. You
do NOT recommend a treatment. You do NOT select a best option. You
describe the comparison, the magnitude, the uncertainty, and the
caveats. The clinician is the decision-maker.

Hard rules:
1. No recommendation language. Never write "should prescribe X",
   "the best choice is Y", "the recommended treatment is Z",
   "the evidence supports starting", or any equivalent.
2. Every magnitude estimate must be accompanied by its confidence
   interval and any flags (out-of-distribution, estimator
   disagreement) must be surfaced explicitly.
3. Required caveats: estimates are conditional-average effects from
   observational data, not individual guarantees; the clinician's
   judgment and patient preferences are essential and not in the model.
4. Reference only the data in the Context. Do not invent treatment
   effect estimates, cohort sizes, or outcome figures.{strictness_addendum}

Context (de-identified):
{json.dumps(de_id, indent=2)}

Return ONLY valid JSON (no prose, no code fences) with this shape:
{{
  "headline":                          "<10-15 words; comparison framing, no recommendation>",
  "comparison_paragraph":              "<3-5 sentence comparison; cite estimates with CIs>",
  "per_treatment_summary": {{
     "<treatment_id>": "<1-2 sentences per treatment; describe context not selection>"
  }},
  "uncertainty_summary":               "<1-2 sentences on uncertainty sources>",
  "caveats":                           ["<caveat>", "<caveat>", "<caveat>", "<caveat>"],
  "suggested_clinician_review_points": ["<point>", "<point>", "<point>"]
}}
"""
    response = bedrock_runtime.invoke_model(
        modelId=COMPARISON_BRIEFING_MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens":        1500,
            "temperature":       0.0,
            "messages":          [{"role": "user", "content": prompt}],
        }),
    )
    payload = json.loads(response["body"].read())
    completion = payload["content"][0]["text"]

    match = re.search(r"\{.*\}", completion, re.DOTALL)
    if not match:
        raise ValueError("LLM returned no JSON object")
    return json.loads(match.group(0))

def _validate_briefing(briefing: dict, observed_context: dict) -> dict:
    """
    Four-layer validator from the main recipe.

    1. Schema and length: required fields present; no oversize text.
    2. Recommendation language: pattern matching for recommendation
       phrasing. This is the strict layer.
    3. Uncertainty completeness: every estimate mentioned must be
       accompanied by its confidence interval reference, and any
       OOD or disagreement flags must be surfaced.
    4. Required caveats: explicit acknowledgment of observational-data
       limitations, conditional-average framing, and clinician-as-
       decision-maker.

    Returns a dict with `passed` (bool) and `feedback` (list of
    issues) for the regeneration loop.
    """
    issues = []

    # Layer 1: schema and length.
    required_keys = {
        "headline", "comparison_paragraph", "per_treatment_summary",
        "uncertainty_summary", "caveats",
        "suggested_clinician_review_points",
    }
    if not isinstance(briefing, dict):
        return {"passed": False, "feedback": ["briefing_not_dict"]}
    if not required_keys.issubset(briefing.keys()):
        missing = required_keys - set(briefing.keys())
        issues.append(f"missing_required_fields: {sorted(missing)}")

    for key in {"headline", "comparison_paragraph",
                  "uncertainty_summary"}:
        val = briefing.get(key, "")
        if not isinstance(val, str) or not val.strip():
            issues.append(f"empty_or_non_string: {key}")
        elif len(val) > 2000:
            issues.append(f"oversize_text: {key}")

    if not isinstance(briefing.get("caveats", []), list):
        issues.append("caveats_not_list")
    if not isinstance(briefing.get("suggested_clinician_review_points", []),
                       list):
        issues.append("review_points_not_list")
    if not isinstance(briefing.get("per_treatment_summary", {}), dict):
        issues.append("per_treatment_summary_not_dict")

    full_text = " ".join(
        v if isinstance(v, str)
        else " ".join(str(x) for x in v) if isinstance(v, list)
        else " ".join(str(x) for x in v.values()) if isinstance(v, dict)
        else str(v)
        for v in briefing.values()
    ).lower()

    # Layer 2: recommendation language.
    recommendation_patterns = [
        r"\bshould prescribe\b",
        r"\bshould be prescribed\b",
        r"\bshould start\b",
        r"\bbest choice\b",
        r"\bbest option\b",
        r"\brecommended treatment\b",
        r"\brecommend (?:starting|prescribing|that)\b",
        r"\bthe evidence supports (?:starting|prescribing)\b",
        r"\bclearly the (?:better|best)\b",
        r"\bsuperior choice\b",
        r"\bdefinitely (?:choose|prescribe)\b",
    ]
    for pattern in recommendation_patterns:
        if re.search(pattern, full_text):
            issues.append(f"recommendation_language_detected: {pattern}")

    # Layer 3: uncertainty completeness. Heuristic: if pair_results
    # contained OOD-flagged or disagreement-flagged entries, the
    # briefing should mention "out-of-distribution", "out of
    # distribution", "wide interval", "uncertainty", or
    # "disagreement" somewhere.
    if observed_context.get("any_oodflag"):
        if not any(term in full_text
                    for term in ["out-of-distribution", "out of distribution",
                                  "limited similar", "extrapolat",
                                  "few comparable"]):
            issues.append("oodflag_present_but_not_surfaced")
    if observed_context.get("any_disagreement"):
        if not any(term in full_text
                    for term in ["disagreement", "estimators disagree",
                                  "wide interval", "wide confidence"]):
            issues.append("disagreement_present_but_not_surfaced")

    # Layer 4: required caveats. The caveats list must include
    # at least one entry referencing observational data and one
    # referencing the clinician's role.
    caveats_text = " ".join(
        c.lower() if isinstance(c, str) else ""
        for c in briefing.get("caveats", [])
    )
    if not any(term in caveats_text
                for term in ["observational", "not randomized",
                              "real-world data"]):
        issues.append("missing_observational_caveat")
    if not any(term in caveats_text
                for term in ["clinician", "clinical judgment",
                              "shared decision", "patient prefer"]):
        issues.append("missing_clinician_judgment_caveat")

    return {
        "passed":   len(issues) == 0,
        "feedback": issues,
    }

def _templated_briefing_fallback(context: dict) -> dict:
    """
    Deterministic fallback briefing when LLM generation or the
    validator fails. Lists the structured comparison without LLM
    narration. Less readable but always passes validation; faithful
    to the data.
    """
    pair_summaries = {}
    headline_tx = []
    for pair in context.get("pair_results", []):
        if pair.get("scoring_status") == "suppressed_oodflag":
            pair_summaries[pair["treatment_pair_id"]] = (
                f"{pair['treatment_id']} vs {pair['comparator_id']}: "
                f"estimate suppressed (OOD severity "
                f"{pair['ood_flag']['severity']:.2f})."
            )
            continue
        pair_summaries[pair["treatment_pair_id"]] = (
            f"{pair['treatment_id']} vs {pair['comparator_id']}: "
            f"estimated {pair['outcome_id']} change "
            f"{pair['point_estimate']:+.2f} (CI {pair['ci_low']:+.2f} "
            f"to {pair['ci_high']:+.2f}); cohort size "
            f"{pair.get('cohort_summary', {}).get('total_size', 'unknown')}."
        )
        headline_tx.append(pair["treatment_id"])

    return {
        "headline": (
            "Structured comparison across "
            f"{len(headline_tx)} treatment(s); briefing fallback used. "
            "Refer to the values below."
        ),
        "comparison_paragraph": (
            "The model produced estimates for the eligible treatment-comparator "
            "pairs listed below. The narrative briefing was not generated "
            "(LLM unavailable or validator rejected output). Estimates are "
            "from observational data with target trial emulation and reflect "
            "conditional-average treatment effects within the matched cohort."
        ),
        "per_treatment_summary": pair_summaries,
        "uncertainty_summary": (
            "Confidence intervals reflect statistical uncertainty plus "
            "estimator agreement plus pre-computed sensitivity-analysis bounds."
        ),
        "caveats": [
            "Estimates are from observational data, not randomized trials.",
            "Estimates are conditional-average effects within the matched cohort, "
              "not individual guarantees.",
            "The clinician's judgment and patient preferences are essential "
              "and not captured in the model.",
            "Patient barriers (cost, injectable acceptance, supply) should "
              "drive shared decision-making.",
        ],
        "suggested_clinician_review_points": [
            "Review the patient's cardiovascular and renal indications.",
            "Confirm formulary status and current copay.",
            "Discuss patient preferences regarding modality and side effects.",
        ],
        "fallback_reason": "templated_briefing_used",
    }

def _no_eligible_briefing(scoring: dict) -> dict:
    """Templated 'no eligible pairs' briefing."""
    return {
        "headline": (
            "No eligible treatment-comparator pairs for this patient at "
            "this index condition."
        ),
        "comparison_paragraph": (
            "The treatment catalog has no production-ready pairs that "
            "match the patient's profile and the requested index condition."
        ),
        "per_treatment_summary":             {},
        "uncertainty_summary":               "no_estimates_to_present",
        "caveats": [
            "This is not a contraindication; it indicates the catalog "
              "does not currently support this decision.",
            "Standard clinical workflow continues. The clinician's "
              "judgment and the institutional formulary apply.",
        ],
        "suggested_clinician_review_points": [
            "Use standard clinical guidelines for this decision.",
            "Consider whether the patient's profile suggests a referral "
              "or specialist consultation.",
        ],
        "validator_status": True,
    }

def _scoring_run_patient(scoring_run_id: str) -> str:
    """Extract the patient_id from a scoring_run_id (demo-only helper)."""
    parts = scoring_run_id.split("-")
    # Format: score-{run_date}-{patient_id}-{suffix}
    if len(parts) >= 4:
        return "-".join(parts[2:-1])
    return ""

def _scan_scoring_result(scoring_run_id: str) -> dict:
    """Demo-only fallback: scan for a scoring_run_id."""
    table = dynamodb.Table(SCORING_RESULTS_TABLE)
    try:
        response = table.scan()
        for item in response.get("Items", []):
            item = _from_decimal(item)
            if item.get("scoring_run_id") == scoring_run_id:
                return item
    except Exception:
        pass
    return {}