Recipe 11.10: Python Implementation Example

pip install boto3

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

import boto3
from botocore.config import Config

# Structured logging. In production, ship JSON-formatted records
# to CloudWatch Logs Insights for cross-call investigation.
# Recruitment-conversation content is research data: prospective-
# participant utterances, eligibility-criterion responses, IRB-
# approved trial information surfaced, FAQ retrievals, prescreen
# results, coordinator-handoff dispositions. Log structural
# metadata only (session_id, trial_id, prospective_participant_id_hash,
# referral_source, intent, tool name, tool latency, tool outcome,
# emergency-flag triggered, escalation reason). Never log raw
# user utterances, never log raw generated responses, never log
# tool arguments containing identifiers or PHI, never log IRB-
# approved-content excerpts. Full transcripts and full tool
# call records live in the research-data audit pipeline (Firehose
# plus Object-Lock S3) under research-grade retention.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from Bedrock, DynamoDB,
# EventBridge, Firehose, CloudWatch, S3, Pinpoint, Connect,
# Step Functions, and Secrets Manager.
BOTO3_RETRY_CONFIG = Config(
    retries={"max_attempts": 3, "mode": "adaptive"})

# Module-level clients. Reused across Lambda invocations in
# warm containers so each invocation does not pay the
# connection cost.
REGION = "us-east-1"
bedrock_runtime       = boto3.client("bedrock-runtime",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
bedrock_agent_runtime = boto3.client("bedrock-agent-runtime",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
dynamodb              = boto3.resource("dynamodb",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
eventbridge_client    = boto3.client("events",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
firehose_client       = boto3.client("firehose",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
cloudwatch_client     = boto3.client("cloudwatch",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
s3_client             = boto3.client("s3",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
secrets_client        = boto3.client("secretsmanager",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
pinpoint_client       = boto3.client("pinpoint",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
# Note: secrets_client and pinpoint_client are declared for
# completeness (the production architecture uses Secrets
# Manager for upstream-system credentials and Pinpoint for
# proactive recruitment messaging). The demo exercises them
# symbolically; production wires them through their
# respective tool-implementation Lambdas.
connect_client        = boto3.client("connect",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)
sfn_client            = boto3.client("stepfunctions",
                                     region_name=REGION,
                                     config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Per-trial isolation is enforced through a partition key
# convention: every trial-specific record carries a
# `trial_id` partition key. The per-trial KMS encryption
# context binds writes to the correct trial.
TRIAL_CONTEXT_TABLE              = "tr-trial-context"
TRIAL_STATE_TABLE                = "tr-trial-state"
ELIGIBILITY_RULE_TABLE           = "tr-eligibility-rules"
RECRUITMENT_FAQ_TABLE            = "tr-recruitment-faq"
PRESCREEN_RECORD_TABLE           = "tr-prescreen-records"
CONVERSATION_STATE_TABLE         = "tr-conversation-state"
CONVERSATION_METADATA_TABLE      = "tr-conversation-metadata"
TOOL_CALL_LEDGER_TABLE           = "tr-tool-call-ledger"
DECISION_RECORD_TABLE            = "tr-recruitment-decision-journal"
REPRESENTATIVENESS_TABLE         = "tr-representativeness"
HANDOFF_QUEUE_STATE_TABLE        = "tr-handoff-queue-state"
CONSENT_RECORD_TABLE             = "tr-consent-record"
RECRUITMENT_EVENT_BUS_NAME       = "tr-recruitment-events-bus"
AUDIT_ARCHIVE_FIREHOSE_NAME      = "tr-audit-archive"
DECISION_RECORD_BUCKET           = "tr-decision-journal"
CONVERSATION_ARCHIVE_BUCKET      = "tr-conversation-archive"
PINPOINT_APPLICATION_ID          = "PINPOINT_APP_PLACEHOLDER"
CONNECT_INSTANCE_ID              = "CONNECT_INSTANCE_PLACEHOLDER"
CONNECT_CONTACT_FLOW_ID          = "CONNECT_FLOW_PLACEHOLDER"
TRIAL_ONBOARDING_SM_ARN          = "TRIAL_ONBOARDING_SM_ARN_PLACEHOLDER"
IRB_AMENDMENT_SM_ARN             = "IRB_AMENDMENT_SM_ARN_PLACEHOLDER"
CLOUDWATCH_NAMESPACE             = "ClinicalTrialRecruitment"

# Bedrock Knowledge Base IDs. The trial-corpus KB is the
# IRB-approved trial-specific content (one or more KBs in
# production, one per trial or one per therapeutic area
# with strict per-trial filtering at retrieval time). The
# pattern KB holds the institutional recruitment-conversation
# pattern library (general patterns layered with per-trial
# variations).
TRIAL_CORPUS_KB_ID               = "TRIAL_KB_PLACEHOLDER"
RECRUITMENT_PATTERN_KB_ID        = "PATTERN_KB_PLACEHOLDER"
FAQ_KB_ID                        = "FAQ_KB_PLACEHOLDER"

# Bedrock Guardrail. Pin to a specific version, not DRAFT,
# in production. The Guardrail's restricted-topic filters
# block recommendation-language, trial-comparison-language,
# clinical-decision attempts, prescription attempts, benefits-
# quote attempts, triage attempts, and therapy attempts.
GUARDRAIL_ID                     = "GUARDRAIL_PLACEHOLDER_ID"
GUARDRAIL_VERSION                = "1"

# KMS key IDs. The recruitment-decision-record journal uses
# a separately-managed customer key for blast-radius
# containment. A leaked credential to the general
# recruitment workload should not give an attacker the
# decision-record archive.
GENERAL_KMS_KEY_ID               = "GENERAL_KMS_PLACEHOLDER"
DECISION_RECORD_KMS_KEY_ID       = "DECISION_KMS_PLACEHOLDER"
CONVERSATION_ARCHIVE_KMS_KEY_ID  = "CONV_ARCHIVE_KMS_PLACEHOLDER"

# Deploy-time guardrail. Any blank resource name is a deploy-
# time bug, not a runtime surprise.
for _name, _value in [
    ("TRIAL_CONTEXT_TABLE",          TRIAL_CONTEXT_TABLE),
    ("TRIAL_STATE_TABLE",            TRIAL_STATE_TABLE),
    ("ELIGIBILITY_RULE_TABLE",       ELIGIBILITY_RULE_TABLE),
    ("RECRUITMENT_FAQ_TABLE",        RECRUITMENT_FAQ_TABLE),
    ("PRESCREEN_RECORD_TABLE",       PRESCREEN_RECORD_TABLE),
    ("CONVERSATION_STATE_TABLE",     CONVERSATION_STATE_TABLE),
    ("CONVERSATION_METADATA_TABLE",  CONVERSATION_METADATA_TABLE),
    ("TOOL_CALL_LEDGER_TABLE",       TOOL_CALL_LEDGER_TABLE),
    ("DECISION_RECORD_TABLE",        DECISION_RECORD_TABLE),
    ("REPRESENTATIVENESS_TABLE",     REPRESENTATIVENESS_TABLE),
    ("HANDOFF_QUEUE_STATE_TABLE",    HANDOFF_QUEUE_STATE_TABLE),
    ("CONSENT_RECORD_TABLE",         CONSENT_RECORD_TABLE),
    ("RECRUITMENT_EVENT_BUS_NAME",   RECRUITMENT_EVENT_BUS_NAME),
    ("AUDIT_ARCHIVE_FIREHOSE_NAME",  AUDIT_ARCHIVE_FIREHOSE_NAME),
    ("DECISION_RECORD_BUCKET",       DECISION_RECORD_BUCKET),
    ("CONVERSATION_ARCHIVE_BUCKET",  CONVERSATION_ARCHIVE_BUCKET),
]:
    if not _value:
        raise RuntimeError(
            f"Resource name {_name} is empty. "
            f"Configure deployment-time values.")

# --- Bedrock Models ---
ORCHESTRATION_MODEL_ID = (
    "anthropic.claude-3-5-sonnet-20241022-v2:0")
PREFILTER_MODEL_ID     = (
    "anthropic.claude-3-5-haiku-20241022-v1:0")
SUMMARY_MODEL_ID       = (
    "anthropic.claude-3-5-haiku-20241022-v1:0")

# --- Conversation tuning ---
MAX_AGENT_ITERATIONS         = 8
TOOL_CALL_TIMEOUT_SECONDS    = 8
RETRIEVAL_TOP_K              = 4
RETRIEVAL_SCORE_FLOOR        = Decimal("0.55")
CITATION_COVERAGE_FLOOR      = Decimal("0.85")

# --- Prescreen disposition codes ---
DISPOSITION_DISQUALIFIED          = "DISQUALIFIED"
DISPOSITION_UNCERTAIN_PENDING     = "UNCERTAIN_PENDING_COORDINATOR"
DISPOSITION_LIKELY_ELIGIBLE       = "LIKELY_ELIGIBLE_PENDING_COORDINATOR"
DISPOSITION_DECLINED_BY_PATIENT   = "DECLINED_BY_PATIENT"
DISPOSITION_TRIAL_CLOSED          = "TRIAL_CLOSED_OR_PAUSED"
DISPOSITION_OUT_OF_SCOPE          = "OUT_OF_SCOPE_ROUTED"
DISPOSITION_EMERGENCY_ROUTED      = "EMERGENCY_ROUTED"

DISPOSITIONS = {
    DISPOSITION_DISQUALIFIED,
    DISPOSITION_UNCERTAIN_PENDING,
    DISPOSITION_LIKELY_ELIGIBLE,
    DISPOSITION_DECLINED_BY_PATIENT,
    DISPOSITION_TRIAL_CLOSED,
    DISPOSITION_OUT_OF_SCOPE,
    DISPOSITION_EMERGENCY_ROUTED,
}

# --- Trial-state codes ---
TRIAL_STATE_OPEN                  = "OPEN_FOR_ENROLLMENT"
TRIAL_STATE_PAUSED                = "ENROLLMENT_PAUSED"
TRIAL_STATE_CLOSED                = "ENROLLMENT_CLOSED"
TRIAL_STATE_AMENDMENT_PENDING     = "AMENDMENT_PENDING_IRB_REVIEW"
TRIAL_STATE_AMENDMENT_REJECTED    = "AMENDMENT_REJECTED"

# --- Identity classification (vulnerable-populations posture) ---
IDENTITY_ADULT_SELF               = "ADULT_SELF_DECISION"
IDENTITY_PARENT_GUARDIAN          = "PARENT_GUARDIAN_PEDIATRIC"
IDENTITY_SURROGATE_DECISION_MAKER = "SURROGATE_DECISION_MAKER"
IDENTITY_PROTECTED_POPULATION_B   = "PROTECTED_POPULATION_PREGNANT"
IDENTITY_PROTECTED_POPULATION_C   = "PROTECTED_POPULATION_PRISONER"
IDENTITY_PROTECTED_POPULATION_D   = "PROTECTED_POPULATION_CHILDREN"

# --- Cohort-stratification dimensions for representativeness ---
# Captured per the trial's IRB-approved data-collection plan
# and the sponsor's diversity action plan where applicable.
# Patient self-report; no demographic inference from name,
# voice, or other indirect signals. Patients can decline
# any demographic question without losing access to the
# recruitment conversation.
COHORT_DIMENSIONS = [
    "language",
    "race_ethnicity",       # OMB categories with self-report
    "sex_gender",
    "age_cohort",
    "geography_zip3",       # Three-digit ZIP for de-identification
    "insurance_category",   # commercial / medicaid / medicare / uninsured
    "referral_source",      # clinician / sponsor / registry / self
    "channel",              # web_chat / sms / voice / portal
    "site_id",              # The trial site routing the recruitment
]

# --- Recruitment funnel stages, per the recruitment plan ---
FUNNEL_STAGE_ENTERED              = "ENTERED"
FUNNEL_STAGE_DISCLOSURE_ACCEPTED  = "DISCLOSURE_ACCEPTED"
FUNNEL_STAGE_FAQ_ENGAGED          = "FAQ_ENGAGED"
FUNNEL_STAGE_PRESCREEN_STARTED    = "PRESCREEN_STARTED"
FUNNEL_STAGE_PRESCREEN_COMPLETED  = "PRESCREEN_COMPLETED"
FUNNEL_STAGE_HANDOFF_SCHEDULED    = "HANDOFF_SCHEDULED"
FUNNEL_STAGE_HANDOFF_ACCEPTED     = "HANDOFF_ACCEPTED_BY_COORDINATOR"
FUNNEL_STAGE_CONSENTED            = "CONSENTED_PER_COORDINATOR_REPORT"
FUNNEL_STAGE_RANDOMIZED           = "RANDOMIZED_PER_COORDINATOR_REPORT"

# --- Emergency-screening signals ---
# Continuous emergency screening across every utterance,
# same architectural primitive as recipes 11.6, 11.7, 11.8,
# and 11.9. The trigger phrases live in a curated, clinical-
# leadership-owned, multilingual rule library. The list
# below is illustrative; production deployments use a
# materially larger rule library reviewed clinically and
# updated continuously.
EMERGENCY_PATTERNS_RAW = [
    r"\bchest pain\b",
    r"\bpressure in (?:my|the) chest\b",
    r"\bcan't breathe\b",
    r"\bcannot breathe\b",
    r"\bshort(?:ness)? of breath\b",
    r"\bnumb(?:ness)? on one side\b",
    r"\bface(?: is)? drooping\b",
    r"\bslurr(?:ed|ing) speech\b",
    r"\bworst headache (?:of|in) my life\b",
    r"\bsuicid(?:e|al)\b",
    r"\bkill myself\b",
    r"\bend my life\b",
    r"\bovers?dose\b",
    r"\babout to (?:hurt|harm) (?:my|some)self\b",
    r"\b911\b",
    r"\bemergency room\b",
]
EMERGENCY_PATTERNS = [
    re.compile(p, re.IGNORECASE) for p in EMERGENCY_PATTERNS_RAW]

# --- Out-of-scope routing rules ---
# Topics outside recruitment scope route to specific
# alternative pathways. The mapping below is illustrative;
# institutional production deployments tailor the routing
# table to their actual care-team and compliance-team
# escalation surface.
OUT_OF_SCOPE_PATTERNS = {
    "clinical_advice_about_existing_care": [
        r"\bshould I (?:take|stop|change) (?:my )?\w+",
        r"\bis (?:it|this) safe for me to\b",
        r"\bdo I have\b.*\b(?:diabetes|cancer|heart attack|stroke)\b",
    ],
    "trial_recommendation_request": [
        r"\bdo you think (?:I should|this trial)\b",
        r"\bwhat would you do\b",
        r"\bwhich trial is best\b",
    ],
    "benefits_quote_request": [
        r"\bhow much will (?:my )?insurance pay\b",
        r"\bwill (?:medicare|medicaid|my insurance) cover\b",
    ],
    "prescription_request": [
        r"\bcan I get a prescription\b",
        r"\brefill my\b",
    ],
}
OUT_OF_SCOPE_COMPILED = {
    category: [re.compile(p, re.IGNORECASE) for p in patterns]
    for category, patterns in OUT_OF_SCOPE_PATTERNS.items()
}

# --- IRB-approved disclosure language identifiers ---
# Each IRB-approved disclosure has a unique identifier that
# the orchestration prompt and the output-safety layer use
# to verify that required disclosures were surfaced. The
# actual text comes from the trial-context store with strict
# version pinning.
DISCLOSURE_ASSISTANT_NOT_PERSON   = "DISC.ASSISTANT_NOT_PERSON"
DISCLOSURE_NOT_COORDINATOR        = "DISC.NOT_COORDINATOR"
DISCLOSURE_CANNOT_ENROLL          = "DISC.CANNOT_ENROLL"
DISCLOSURE_PROVIDING_INFO_ONLY    = "DISC.PROVIDING_INFO_ONLY"
DISCLOSURE_CAN_STOP_ANY_TIME      = "DISC.CAN_STOP_ANY_TIME"
DISCLOSURE_DATA_RETENTION_NOTICE  = "DISC.DATA_RETENTION_NOTICE"
DISCLOSURE_REQUEST_COORDINATOR    = "DISC.REQUEST_COORDINATOR_ANY_TIME"
REQUIRED_DISCLOSURES_FIRST_TURN = [
    DISCLOSURE_ASSISTANT_NOT_PERSON,
    DISCLOSURE_NOT_COORDINATOR,
    DISCLOSURE_CANNOT_ENROLL,
    DISCLOSURE_PROVIDING_INFO_ONLY,
    DISCLOSURE_CAN_STOP_ANY_TIME,
    DISCLOSURE_DATA_RETENTION_NOTICE,
    DISCLOSURE_REQUEST_COORDINATOR,
]

# --- Eligibility-criterion category taxonomy ---
# Each eligibility criterion is classified into one of the
# four categories below. The category determines how the
# rule engine evaluates the criterion and what the assistant
# is permitted to say about the result.
CRITERION_CATEGORY_SIMPLE_STRUCTURED   = "SIMPLE_STRUCTURED"
CRITERION_CATEGORY_COMPLEX_STRUCTURED  = "COMPLEX_STRUCTURED"
CRITERION_CATEGORY_CLINICAL_JUDGMENT   = "CLINICAL_JUDGMENT"
CRITERION_CATEGORY_VERIFICATION_ONLY   = "VERIFICATION_ONLY"

# --- Eligibility evaluation outcomes (per criterion) ---
EVAL_OUTCOME_MET                        = "MET"
EVAL_OUTCOME_NOT_MET                    = "NOT_MET"
EVAL_OUTCOME_INDETERMINATE              = "INDETERMINATE"
EVAL_OUTCOME_REQUIRES_COORDINATOR       = "REQUIRES_COORDINATOR_REVIEW"

# --- PHI redaction patterns for log lines ---
# Defensive redaction. Production deployments combine pattern-
# based redaction with a managed PII-detection service for
# higher-confidence redaction.
PHI_PATTERNS = {
    "ssn":       re.compile(r"\b\d{3}[-\s]?\d{2}[-\s]?\d{4}\b"),
    "phone":     re.compile(r"\b\d{3}[-\s.]?\d{3}[-\s.]?\d{4}\b"),
    "email":     re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"),
    "mrn":       re.compile(r"\bMRN[:\s]*\d+\b", re.IGNORECASE),
    "dob":       re.compile(r"\b\d{1,2}/\d{1,2}/(?:19|20)\d{2}\b"),
    "zip5":      re.compile(r"\b\d{5}(?:-\d{4})?\b"),
    # Note: the zip5 pattern over-matches (any 5-digit
    # number). Production combines pattern-based redaction
    # with a managed PII-detection service (Comprehend or
    # Macie) for contextual precision. The pattern is
    # acceptable for log-line redaction where false
    # positives are preferable to false negatives.
    "name_hint": re.compile(
        r"\bmy name is\s+([A-Z][a-z]+\s+[A-Z][a-z]+)",
        re.IGNORECASE),
}

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

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

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

def _now() -> datetime:
    """Current UTC datetime."""
    return datetime.now(timezone.utc)

def _redact_pii_for_logging(text: str) -> str:
    """Light redaction for log lines."""
    redacted = text
    for pattern in PHI_PATTERNS.values():
        redacted = pattern.sub("[REDACTED]", redacted)
    return redacted

def _hash_id(raw_id: str) -> str:
    """
    Deterministic, salted hash for log-safe identifiers.
    Production uses a separately-managed pepper retrieved
    from Secrets Manager. The placeholder below makes the
    helper runnable for the demo without hiding that the
    real implementation has additional structure.
    """
    import hashlib
    pepper = b"DEMO_PEPPER_REPLACE_IN_PRODUCTION"
    return hashlib.sha256(pepper + raw_id.encode("utf-8")).hexdigest()[:16]

def _emit_event(detail_type: str, detail: dict) -> None:
    """
    Emit an EventBridge event. Wrapped in try/except so a
    transient EventBridge failure does not block the
    pipeline. Production downstream consumers use the
    suggested idempotency keys (session_id, prescreen_id,
    handoff_id, decision_id, trial_id+amendment_version)
    to deduplicate on retry.
    """
    try:
        eventbridge_client.put_events(Entries=[{
            "Source":       "clinical_trial_recruitment",
            "DetailType":   detail_type,
            "Detail":       json.dumps(_from_decimal(detail)),
            "EventBusName": RECRUITMENT_EVENT_BUS_NAME,
        }])
    except Exception as exc:
        logger.error(
            "EventBridge emit failed for %s: %s",
            detail_type, exc)

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

def _audit_tool_call(session_id: str,
                      trial_id: str,
                      tool: str,
                      arguments: dict,
                      result_summary: dict,
                      latency_ms: int,
                      outcome: str) -> None:
    """Persist a tool-call ledger entry."""
    try:
        table = dynamodb.Table(TOOL_CALL_LEDGER_TABLE)
        table.put_item(Item=_to_decimal({
            "session_id":         session_id,
            "invoked_at":         _now_iso(),
            "trial_id":           trial_id,
            "tool":               tool,
            "arguments_summary":  _redact_tool_args(arguments),
            "result_summary":     result_summary,
            "latency_ms":         latency_ms,
            "outcome":            outcome,
        }))
    except Exception as exc:
        logger.error(
            "Tool-call ledger write failed for %s/%s: %s",
            session_id, tool, exc)

def _redact_tool_args(arguments: dict) -> dict:
    """Strip sensitive fields before ledger storage."""
    redacted = dict(arguments)
    sensitive_keys = {
        "patient_id", "prospective_participant_id",
        "name", "date_of_birth", "user_message",
        "free_text", "phone", "email", "address",
        "guardian_name", "guardian_phone",
        "surrogate_name", "surrogate_phone",
        "irb_corpus_excerpt", "transcript",
    }
    for key in list(redacted.keys()):
        if key in sensitive_keys:
            redacted[key] = "[REDACTED]"
    return redacted

def _detect_emergency_signal(text: str) -> Optional[dict]:
    """
    Continuous emergency screening across every utterance.
    Returns a structured signal dict if any emergency
    pattern fires; otherwise None. Production deployments
    layer the regex prefilter with a tuned classifier and
    a clinical-leadership-owned rule library reviewed
    monthly. The regex set here is illustrative.
    """
    for pattern in EMERGENCY_PATTERNS:
        match = pattern.search(text)
        if match:
            return {
                "matched_pattern":  pattern.pattern,
                "matched_text":     "[REDACTED]",  # do not log raw
                "detected_at":      _now_iso(),
                "severity":         "ACUTE_EMERGENCY",
            }
    return None

def _detect_out_of_scope(text: str) -> Optional[str]:
    """
    Return the out-of-scope category name if any out-of-
    scope pattern matches; otherwise None. The first match
    wins; production deployments typically score multiple
    matches and route to the most severe category.
    """
    for category, patterns in OUT_OF_SCOPE_COMPILED.items():
        for pattern in patterns:
            if pattern.search(text):
                return category
    return None

def _shape_session_id(prefix: str = "tr_session") -> str:
    """Generate a session identifier with a recognizable prefix."""
    return f"{prefix}_{uuid.uuid4().hex[:24]}"

def _shape_decision_id(prefix: str = "tr_decision") -> str:
    """Generate a recruitment-decision identifier."""
    return f"{prefix}_{uuid.uuid4().hex[:24]}"

def _shape_handoff_id(prefix: str = "tr_handoff") -> str:
    """Generate a coordinator-handoff identifier."""
    return f"{prefix}_{uuid.uuid4().hex[:24]}"

def _safe_get(table_name: str, key: dict) -> Optional[dict]:
    """
    Defensive DynamoDB GetItem wrapper. Returns the Item
    dict if present, None on miss, raises only on transient
    failures left for the caller to handle.
    """
    try:
        table = dynamodb.Table(table_name)
        response = table.get_item(Key=key)
        return response.get("Item")
    except Exception as exc:
        logger.error(
            "DynamoDB GetItem failed for %s key=%s: %s",
            table_name, key, exc)
        raise

class MockTable:
    """In-memory DynamoDB stand-in. Single-table, single-key
    semantics; no GSI emulation. Production deployments use
    real DynamoDB with appropriately-configured GSIs for the
    representativeness, conversation-metadata, and tool-
    call-ledger tables."""

    def __init__(self, name: str, key_attr: str,
                 sort_attr: Optional[str] = None):
        self.name = name
        self.key_attr = key_attr
        self.sort_attr = sort_attr
        self._store: dict = {}

    def _composite_key(self, key: dict) -> tuple:
        if self.sort_attr is not None:
            return (key[self.key_attr], key[self.sort_attr])
        return (key[self.key_attr],)

    def put_item(self, Item: dict, **_kwargs):
        if self.sort_attr is not None:
            ck = (Item[self.key_attr], Item[self.sort_attr])
        else:
            ck = (Item[self.key_attr],)
        self._store[ck] = dict(Item)
        return {}

    def get_item(self, Key: dict, **_kwargs):
        ck = self._composite_key(Key)
        item = self._store.get(ck)
        return {"Item": item} if item is not None else {}

    def update_item(self, Key, UpdateExpression, ExpressionAttributeValues,
                    ExpressionAttributeNames=None, **_kwargs):
        # Simplified UpdateExpression handler for the demo.
        # Real DynamoDB supports SET / ADD / REMOVE / DELETE
        # with conditional expressions and atomic counters.
        ck = self._composite_key(Key)
        item = dict(self._store.get(ck, dict(Key)))
        # Parse very light SET assignments: "SET a = :a, b = :b"
        if UpdateExpression.startswith("SET "):
            assignments = UpdateExpression[4:].split(",")
            for assignment in assignments:
                target, value_ref = [
                    p.strip() for p in assignment.split("=", 1)]
                # Resolve attribute name aliases.
                if (ExpressionAttributeNames
                        and target.startswith("#")):
                    target = ExpressionAttributeNames[target]
                item[target] = ExpressionAttributeValues[value_ref]
        self._store[ck] = item
        return {"Attributes": item}

    def query(self, KeyConditionExpression=None,
              ExpressionAttributeValues=None,
              ExpressionAttributeNames=None,
              FilterExpression=None,
              **_kwargs):
        # Demo-grade: returns all items for this table that
        # match the partition key in the expression. Real
        # query semantics are richer.
        results = list(self._store.values())
        return {"Items": results, "Count": len(results)}

class MockEventBus:
    def __init__(self):
        self.events: list = []

    def put_events(self, Entries):
        self.events.extend(Entries)
        return {"FailedEntryCount": 0,
                "Entries": [{"EventId": uuid.uuid4().hex}
                             for _ in Entries]}

class MockCloudWatch:
    def __init__(self):
        self.metrics: list = []

    def put_metric_data(self, Namespace, MetricData):
        for m in MetricData:
            self.metrics.append({"namespace": Namespace, **m})
        return {}

class MockS3:
    def __init__(self):
        self.objects: dict = {}

    def put_object(self, Bucket, Key, Body, **_kwargs):
        self.objects[(Bucket, Key)] = Body
        return {"ETag": uuid.uuid4().hex}

class MockPinpoint:
    def __init__(self):
        self.messages: list = []

    def send_messages(self, ApplicationId, MessageRequest):
        self.messages.append({"app": ApplicationId,
                              "request": MessageRequest})
        return {"MessageResponse": {"Result": {}}}

class MockConnect:
    def __init__(self):
        self.contacts: list = []

    def start_chat_contact(self, **kwargs):
        contact_id = uuid.uuid4().hex
        self.contacts.append({"contact_id": contact_id, **kwargs})
        return {"ContactId": contact_id,
                "ParticipantId": uuid.uuid4().hex,
                "ParticipantToken": uuid.uuid4().hex}

class MockStepFunctions:
    def __init__(self):
        self.executions: list = []

    def start_execution(self, stateMachineArn, name, input):
        execution_arn = (
            f"{stateMachineArn}:execution:{name}")
        self.executions.append({
            "arn": execution_arn,
            "input": input})
        return {"executionArn": execution_arn,
                "startDate": _now()}

class MockBedrockAgentRuntime:
    """Stand-in for the Bedrock Knowledge Base retrieval
    surface. Returns a fixed corpus for the demo."""

    def __init__(self, kb_corpora):
        self._kb_corpora = kb_corpora

    def retrieve(self, knowledgeBaseId, retrievalQuery,
                 retrievalConfiguration=None):
        corpus = self._kb_corpora.get(knowledgeBaseId, [])
        # Score by simple substring match for the demo.
        query_text = retrievalQuery["text"].lower()
        scored = []
        for entry in corpus:
            text = entry["content"].lower()
            score = sum(
                1 for token in query_text.split()
                if token in text) / max(1, len(query_text.split()))
            scored.append((score, entry))
        scored.sort(reverse=True, key=lambda x: x[0])
        top = scored[:RETRIEVAL_TOP_K]
        return {
            "retrievalResults": [
                {
                    "content": {"text": entry["content"]},
                    "score":   float(score),
                    "location": {
                        "s3Location": {"uri": entry.get(
                            "uri",
                            f"s3://{TRIAL_CORPUS_KB_ID}/"
                            f"{entry['id']}")},
                    },
                    "metadata": entry.get("metadata", {}),
                }
                for score, entry in top if score > 0.0
            ]
        }

class MockBedrockRuntime:
    """Very small stand-in for bedrock-runtime.invoke_model.
    Returns canned tool-call sequences for the demo."""

    def __init__(self):
        self._scripted_responses: list = []

    def queue_response(self, response: dict) -> None:
        """Queue the next canned model response."""
        self._scripted_responses.append(response)

    def invoke_model(self, modelId, body, **_kwargs):
        # Pop the next scripted response. If nothing is
        # queued, return a default "I'd like to connect
        # you to a coordinator" answer with the required
        # disclosures.
        if self._scripted_responses:
            response = self._scripted_responses.pop(0)
        else:
            response = {
                "stop_reason": "end_turn",
                "content": [{
                    "type": "text",
                    "text": (
                        "I appreciate your interest. I'd like "
                        "to connect you with a research "
                        "coordinator who can answer your "
                        "remaining questions and walk you "
                        "through next steps."),
                    "citations": [],
                }],
            }
        return {
            "body": _MockStreamBody(json.dumps(response)),
            "ResponseMetadata": {"HTTPStatusCode": 200},
        }

class _MockStreamBody:
    def __init__(self, payload: str):
        self._payload = payload

    def read(self):
        return self._payload.encode("utf-8")

# Build mock tables and replace clients.
mock_tables = {
    TRIAL_CONTEXT_TABLE:         MockTable(
        TRIAL_CONTEXT_TABLE,        "trial_id"),
    TRIAL_STATE_TABLE:           MockTable(
        TRIAL_STATE_TABLE,          "trial_id"),
    ELIGIBILITY_RULE_TABLE:      MockTable(
        ELIGIBILITY_RULE_TABLE,     "trial_id",
                                    "criterion_id"),
    RECRUITMENT_FAQ_TABLE:       MockTable(
        RECRUITMENT_FAQ_TABLE,      "trial_id",
                                    "faq_id"),
    PRESCREEN_RECORD_TABLE:      MockTable(
        PRESCREEN_RECORD_TABLE,     "session_id"),
    CONVERSATION_STATE_TABLE:    MockTable(
        CONVERSATION_STATE_TABLE,   "session_id"),
    CONVERSATION_METADATA_TABLE: MockTable(
        CONVERSATION_METADATA_TABLE, "session_id"),
    TOOL_CALL_LEDGER_TABLE:      MockTable(
        TOOL_CALL_LEDGER_TABLE,     "session_id",
                                    "invoked_at"),
    DECISION_RECORD_TABLE:       MockTable(
        DECISION_RECORD_TABLE,      "decision_id"),
    REPRESENTATIVENESS_TABLE:    MockTable(
        REPRESENTATIVENESS_TABLE,   "trial_id",
                                    "session_id"),
    HANDOFF_QUEUE_STATE_TABLE:   MockTable(
        HANDOFF_QUEUE_STATE_TABLE,  "handoff_id"),
    CONSENT_RECORD_TABLE:        MockTable(
        CONSENT_RECORD_TABLE,       "session_id"),
}

def _mock_dynamodb_table(name):
    return mock_tables[name]

# Replace boto3 dynamodb.Table with the mock for the demo.
dynamodb.Table = _mock_dynamodb_table

# Replace EventBridge, Pinpoint, S3, CloudWatch, Step
# Functions, Connect, and Bedrock clients with mocks for
# the demo.
eventbridge_client    = MockEventBus()
pinpoint_client       = MockPinpoint()
s3_client             = MockS3()
cloudwatch_client     = MockCloudWatch()
sfn_client            = MockStepFunctions()
connect_client        = MockConnect()
bedrock_runtime       = MockBedrockRuntime()

# The Bedrock Knowledge Base mock loads a small synthetic
# corpus per-trial and per-FAQ. Production replaces this
# with the real bedrock-agent-runtime client.
_KB_CORPUS = {
    TRIAL_CORPUS_KB_ID: [],
    FAQ_KB_ID:          [],
    RECRUITMENT_PATTERN_KB_ID: [],
}
bedrock_agent_runtime = MockBedrockAgentRuntime(_KB_CORPUS)

def onboard_trial(*,
                   trial_id,
                   protocol_identifier,
                   irb_protocol_number,
                   sponsor_name,
                   principal_investigator,
                   therapeutic_area,
                   irb_approval_record,
                   protocol_summary_irb_approved,
                   visit_schedule_irb_approved,
                   study_procedures_irb_approved,
                   sponsor_information_irb_approved,
                   contact_information_irb_approved,
                   diversity_action_plan=None,
                   languages_supported=None,
                   sites=None,
                   identity_posture=None):
    """
    Persist the per-trial IRB-approved context. Each
    content block is pinned to the IRB-approval record
    that authorized it. Subsequent runtime retrievals are
    bounded to the latest IRB-approved version.

    The function is the production entry point for the
    trial-onboarding Step Functions workflow. In production
    each block above is the output of an upstream content-
    authoring workflow with named clinical-leadership and
    IRB-coordinator signoff and version pinning to the IRB
    approval record.
    """
    if languages_supported is None:
        languages_supported = ["en"]
    if sites is None:
        sites = []
    if identity_posture is None:
        identity_posture = [IDENTITY_ADULT_SELF]

    record = {
        "trial_id":                       trial_id,
        "protocol_identifier":            protocol_identifier,
        "irb_protocol_number":            irb_protocol_number,
        "sponsor_name":                   sponsor_name,
        "principal_investigator":         principal_investigator,
        "therapeutic_area":               therapeutic_area,
        "irb_approval_record":            irb_approval_record,
        "protocol_summary_irb_approved":  protocol_summary_irb_approved,
        "visit_schedule_irb_approved":    visit_schedule_irb_approved,
        "study_procedures_irb_approved":  study_procedures_irb_approved,
        "sponsor_information_irb_approved": sponsor_information_irb_approved,
        "contact_information_irb_approved": contact_information_irb_approved,
        "diversity_action_plan":          diversity_action_plan or {},
        "languages_supported":            languages_supported,
        "sites":                          sites,
        "identity_posture":               identity_posture,
        "onboarded_at":                   _now_iso(),
        "content_version":                irb_approval_record.get(
            "content_version", "v1"),
    }

    table = dynamodb.Table(TRIAL_CONTEXT_TABLE)
    table.put_item(Item=_to_decimal(record))

    # Initial trial-state. Most trials start in
    # AMENDMENT_PENDING_IRB_REVIEW until the first
    # post-onboarding IRB-application has approved the
    # recruitment material; the demo simplifies by
    # marking trials OPEN_FOR_ENROLLMENT immediately.
    set_trial_state(
        trial_id=trial_id,
        state=TRIAL_STATE_OPEN,
        amendment_version=record["content_version"],
        reason="Initial onboarding")

    _emit_event("RecruitmentEvent.TrialOnboarded", {
        "trial_id":            trial_id,
        "irb_protocol_number": irb_protocol_number,
        "content_version":     record["content_version"],
        "onboarded_at":        record["onboarded_at"],
    })

    logger.info(
        "Onboarded trial trial_id=%s irb_protocol_number=%s "
        "content_version=%s",
        trial_id, irb_protocol_number,
        record["content_version"])
    return record

def set_trial_state(*, trial_id, state, amendment_version,
                     reason):
    """
    Set the runtime trial-state. The conversation engine
    retrieves this on every turn; closed-or-paused trials
    are routed to the appropriate alternative pathway
    rather than continued under stale assumptions.
    """
    record = {
        "trial_id":          trial_id,
        "state":             state,
        "amendment_version": amendment_version,
        "reason":            reason,
        "set_at":            _now_iso(),
    }
    table = dynamodb.Table(TRIAL_STATE_TABLE)
    table.put_item(Item=_to_decimal(record))
    _put_metric("TrialStateChange", 1, {
        "trial_id":      trial_id,
        "state":         state,
        "amendment_ver": amendment_version,
    })

def register_eligibility_criterion(*,
                                    trial_id,
                                    criterion_id,
                                    description_irb_approved,
                                    category,
                                    rule_definition,
                                    clinical_owner,
                                    irb_review_record,
                                    inclusion=True,
                                    rule_version="v1"):
    """
    Register a single eligibility criterion. The rule
    definition is a deterministic structure interpretable
    by the rule engine. The LLM does not interpret rule
    definitions; it surfaces the question the rule needs
    answered, captures the patient's response, and routes
    the response to the deterministic evaluator.
    """
    if category not in (
        CRITERION_CATEGORY_SIMPLE_STRUCTURED,
        CRITERION_CATEGORY_COMPLEX_STRUCTURED,
        CRITERION_CATEGORY_CLINICAL_JUDGMENT,
        CRITERION_CATEGORY_VERIFICATION_ONLY,
    ):
        raise ValueError(
            f"Unknown criterion category: {category}")

    record = {
        "trial_id":                  trial_id,
        "criterion_id":              criterion_id,
        "description_irb_approved":  description_irb_approved,
        "category":                  category,
        "rule_definition":           rule_definition,
        "clinical_owner":            clinical_owner,
        "irb_review_record":         irb_review_record,
        "inclusion":                 inclusion,
        "rule_version":              rule_version,
        "registered_at":             _now_iso(),
    }
    table = dynamodb.Table(ELIGIBILITY_RULE_TABLE)
    table.put_item(Item=_to_decimal(record))
    return record

def register_recruitment_faq(*,
                              trial_id,
                              faq_id,
                              question_irb_approved,
                              answer_irb_approved,
                              category,
                              irb_review_record,
                              languages=None):
    """
    Register an IRB-approved recruitment-FAQ entry. Each
    answer lives only as IRB-approved text; the runtime
    retrieval surface returns the IRB-approved answer
    verbatim with citation. The LLM is permitted to
    surface the answer (with the patient's question
    context for selection) but not to generate novel
    answer text.
    """
    if languages is None:
        languages = ["en"]
    record = {
        "trial_id":               trial_id,
        "faq_id":                 faq_id,
        "question_irb_approved":  question_irb_approved,
        "answer_irb_approved":    answer_irb_approved,
        "category":               category,
        "irb_review_record":      irb_review_record,
        "languages":              languages,
        "registered_at":          _now_iso(),
    }
    table = dynamodb.Table(RECRUITMENT_FAQ_TABLE)
    table.put_item(Item=_to_decimal(record))

    # Add to the Knowledge Base mock corpus too. In
    # production the IRB-approved-FAQ corpus lands in S3
    # and the Knowledge Base ingestion pipeline runs the
    # embedding job; the trial_id metadata is the per-
    # trial isolation primitive at retrieval time.
    _KB_CORPUS[FAQ_KB_ID].append({
        "id":      f"{trial_id}::{faq_id}",
        "content": (
            f"Q: {question_irb_approved}\n"
            f"A: {answer_irb_approved}"),
        "uri":     f"s3://irb-approved/{trial_id}/{faq_id}.md",
        "metadata": {
            "trial_id":              trial_id,
            "faq_id":                faq_id,
            "category":              category,
            "irb_review_record":     irb_review_record,
        },
    })
    return record

def register_irb_approved_corpus_excerpt(*,
                                           trial_id,
                                           excerpt_id,
                                           content,
                                           section,
                                           irb_review_record):
    """
    Register an IRB-approved trial-content excerpt for the
    trial-corpus Knowledge Base. The runtime retrieval
    surface returns excerpts with citation; the LLM is
    permitted to surface excerpts but not to paraphrase
    them in ways that diverge from the IRB-approved text.
    The output-safety layer enforces this further.
    """
    _KB_CORPUS[TRIAL_CORPUS_KB_ID].append({
        "id":      f"{trial_id}::{excerpt_id}",
        "content": content,
        "uri":     f"s3://irb-approved/{trial_id}/{section}/{excerpt_id}.md",
        "metadata": {
            "trial_id":          trial_id,
            "excerpt_id":        excerpt_id,
            "section":           section,
            "irb_review_record": irb_review_record,
        },
    })

def get_trial_state(trial_id: str) -> dict:
    """
    Retrieve the current trial-state. Production callers
    cache aggressively (per-Lambda invocation, short TTL)
    because every conversation turn calls this. The
    conversation engine should fail-closed on a missing
    trial-state record (route to coordinator) rather than
    fail-open under unknown state.
    """
    record = _safe_get(TRIAL_STATE_TABLE, {"trial_id": trial_id})
    if not record:
        # Fail-closed: missing state means the trial is not
        # known to the recruitment runtime. Production routes
        # this to the coordinator pathway with a flag for the
        # operations team to investigate.
        return {
            "trial_id":          trial_id,
            "state":             TRIAL_STATE_CLOSED,
            "amendment_version": "unknown",
            "reason":            "Trial-state record missing",
            "set_at":            _now_iso(),
        }
    return _from_decimal(record)

def get_trial_context(trial_id: str) -> Optional[dict]:
    """
    Retrieve the per-trial IRB-approved context. Production
    callers cache aggressively per session (the trial-
    context does not change mid-conversation for the same
    amendment_version); a state change mid-conversation
    triggers a re-fetch and re-presentation handler.
    """
    record = _safe_get(TRIAL_CONTEXT_TABLE,
                        {"trial_id": trial_id})
    return _from_decimal(record) if record else None

def request_irb_amendment(*,
                            trial_id,
                            amendment_record,
                            requested_changes,
                            requested_by,
                            irb_application_id):
    """
    Submit an IRB amendment request. In production this is
    a Step Functions workflow that gathers signoffs (named
    clinical-leadership, sponsor regulatory, IRB
    coordinator), updates the IRB application file, and
    only flips the trial-state to AMENDMENT_PENDING_IRB_REVIEW
    on the IRB-coordinator-confirmed submission event. The
    amendment is *not* runtime until the IRB approves it
    and the new content_version is published.
    """
    # Mark trial as pending amendment review.
    set_trial_state(
        trial_id=trial_id,
        state=TRIAL_STATE_AMENDMENT_PENDING,
        amendment_version=amendment_record.get("version"),
        reason=(
            f"Amendment {amendment_record.get('version')} "
            f"submitted via IRB application "
            f"{irb_application_id}"))

    # Kick off Step Functions workflow.
    sfn_client.start_execution(
        stateMachineArn=IRB_AMENDMENT_SM_ARN,
        name=f"amendment-{trial_id}-{amendment_record.get('version')}",
        input=json.dumps({
            "trial_id":            trial_id,
            "amendment_record":    amendment_record,
            "requested_changes":   requested_changes,
            "requested_by":        requested_by,
            "irb_application_id":  irb_application_id,
        }))

    _emit_event("RecruitmentEvent.IRBAmendmentRequested", {
        "trial_id":            trial_id,
        "amendment_version":   amendment_record.get("version"),
        "irb_application_id":  irb_application_id,
        "requested_at":        _now_iso(),
    })

def apply_irb_amendment_approval(*,
                                  trial_id,
                                  amendment_version,
                                  irb_approval_record,
                                  updated_protocol_summary=None,
                                  updated_eligibility_criteria=None,
                                  updated_faq_entries=None,
                                  updated_corpus_excerpts=None,
                                  updated_visit_schedule=None,
                                  updated_study_procedures=None):
    """
    Apply an IRB-approved amendment to the per-trial
    runtime stores. Each updated content block is written
    with the new amendment_version pinned. In-flight
    conversations whose conversation-state references the
    prior amendment_version receive an IRB-approved-
    process-required re-presentation event; the demo
    surfaces the event mechanically without implementing
    the full re-presentation flow.
    """
    # Update trial-context with the new amendment version.
    context = get_trial_context(trial_id)
    if not context:
        raise RuntimeError(
            f"Cannot apply amendment to unknown trial {trial_id}")

    if updated_protocol_summary is not None:
        context["protocol_summary_irb_approved"] = (
            updated_protocol_summary)
    if updated_visit_schedule is not None:
        context["visit_schedule_irb_approved"] = (
            updated_visit_schedule)
    if updated_study_procedures is not None:
        context["study_procedures_irb_approved"] = (
            updated_study_procedures)

    context["irb_approval_record"] = irb_approval_record
    context["content_version"] = amendment_version

    table = dynamodb.Table(TRIAL_CONTEXT_TABLE)
    table.put_item(Item=_to_decimal(context))

    # Update eligibility rules.
    for criterion in (updated_eligibility_criteria or []):
        register_eligibility_criterion(
            trial_id=trial_id,
            criterion_id=criterion["criterion_id"],
            description_irb_approved=criterion["description_irb_approved"],
            category=criterion["category"],
            rule_definition=criterion["rule_definition"],
            clinical_owner=criterion["clinical_owner"],
            irb_review_record=irb_approval_record,
            inclusion=criterion.get("inclusion", True),
            rule_version=amendment_version)

    # Update FAQ corpus.
    for faq in (updated_faq_entries or []):
        register_recruitment_faq(
            trial_id=trial_id,
            faq_id=faq["faq_id"],
            question_irb_approved=faq["question_irb_approved"],
            answer_irb_approved=faq["answer_irb_approved"],
            category=faq.get("category", "general"),
            irb_review_record=irb_approval_record,
            languages=faq.get("languages", ["en"]))

    # Update corpus excerpts.
    for excerpt in (updated_corpus_excerpts or []):
        register_irb_approved_corpus_excerpt(
            trial_id=trial_id,
            excerpt_id=excerpt["excerpt_id"],
            content=excerpt["content"],
            section=excerpt["section"],
            irb_review_record=irb_approval_record)

    # Flip trial-state back to OPEN with the new version.
    set_trial_state(
        trial_id=trial_id,
        state=TRIAL_STATE_OPEN,
        amendment_version=amendment_version,
        reason=(
            f"Amendment {amendment_version} approved by IRB "
            f"({irb_approval_record.get('record_id')})"))

    _emit_event("RecruitmentEvent.IRBAmendmentApproved", {
        "trial_id":            trial_id,
        "amendment_version":   amendment_version,
        "irb_approval_record": irb_approval_record,
        "applied_at":          _now_iso(),
    })

def pause_trial_enrollment(*, trial_id, reason):
    """Pause enrollment. The conversation engine will route
    in-flight conversations and new entries appropriately."""
    set_trial_state(
        trial_id=trial_id,
        state=TRIAL_STATE_PAUSED,
        amendment_version=get_trial_state(trial_id).get(
            "amendment_version", "unknown"),
        reason=reason)

def close_trial_enrollment(*, trial_id, reason):
    """Close enrollment. Subsequent recruitment-conversation
    entries route to the appropriate alternative pathway."""
    set_trial_state(
        trial_id=trial_id,
        state=TRIAL_STATE_CLOSED,
        amendment_version=get_trial_state(trial_id).get(
            "amendment_version", "unknown"),
        reason=reason)

def receive_conversation_turn(*,
                                session_id,
                                trial_id,
                                user_message,
                                referral_source,
                                channel,
                                language="en",
                                identity_context=None,
                                turn_index=0):
    """
    Run the per-turn input pipeline. Returns a dict with
    routing instructions; the chat handler uses the result
    to decide whether to call the agent's tool-use loop or
    to take a non-LLM path (emergency, out-of-scope,
    closed-trial).
    """
    if identity_context is None:
        identity_context = {"identity": IDENTITY_ADULT_SELF}

    # 1. Continuous emergency screening. Runs first; the
    #    LLM does not see emergency utterances.
    emergency_signal = _detect_emergency_signal(user_message)
    if emergency_signal is not None:
        _emit_event("RecruitmentEvent.EmergencyDetected", {
            "session_id":     session_id,
            "trial_id":       trial_id,
            "matched_pattern": emergency_signal["matched_pattern"],
            "detected_at":    emergency_signal["detected_at"],
        })
        _put_metric("EmergencyDetected", 1, {
            "trial_id":       trial_id,
            "channel":        channel,
            "language":       language,
        })
        return {
            "routing":      "EMERGENCY",
            "signal":       emergency_signal,
            "session_id":   session_id,
            "trial_id":     trial_id,
            "turn_index":   turn_index,
        }

    # 2. Out-of-scope detection. Topics outside recruitment
    #    scope route to the appropriate alternative pathway.
    out_of_scope_category = _detect_out_of_scope(user_message)
    if out_of_scope_category is not None:
        _emit_event("RecruitmentEvent.OutOfScopeDetected", {
            "session_id":     session_id,
            "trial_id":       trial_id,
            "category":       out_of_scope_category,
            "detected_at":    _now_iso(),
        })
        _put_metric("OutOfScopeDetected", 1, {
            "trial_id":       trial_id,
            "category":       out_of_scope_category,
        })
        return {
            "routing":     "OUT_OF_SCOPE",
            "category":    out_of_scope_category,
            "session_id":  session_id,
            "trial_id":    trial_id,
            "turn_index":  turn_index,
        }

    # 3. Trial-state retrieval. A closed-or-paused trial
    #    routes the conversation to the alternative pathway
    #    rather than continuing under stale assumptions.
    trial_state = get_trial_state(trial_id)
    if trial_state["state"] in (
        TRIAL_STATE_CLOSED, TRIAL_STATE_PAUSED,
        TRIAL_STATE_AMENDMENT_PENDING,
        TRIAL_STATE_AMENDMENT_REJECTED,
    ):
        _emit_event("RecruitmentEvent.TrialUnavailable", {
            "session_id":  session_id,
            "trial_id":    trial_id,
            "trial_state": trial_state["state"],
            "reason":      trial_state.get("reason"),
            "detected_at": _now_iso(),
        })
        return {
            "routing":     "TRIAL_UNAVAILABLE",
            "trial_state": trial_state,
            "session_id":  session_id,
            "trial_id":    trial_id,
            "turn_index":  turn_index,
        }

    # 4. Trial-context loading. The IRB-approved trial
    #    context is the per-trial conversation backbone.
    trial_context = get_trial_context(trial_id)
    if not trial_context:
        # Production fail-closed: the runtime cannot serve
        # a trial it does not have onboarded content for.
        return {
            "routing":     "TRIAL_UNAVAILABLE",
            "trial_state": trial_state,
            "reason":      "Trial-context record missing",
            "session_id":  session_id,
            "trial_id":    trial_id,
            "turn_index":  turn_index,
        }

    # 5. Identity classification. Adult-self-decision is
    #    the modal scenario. Pediatric, surrogate-decision-
    #    maker, and vulnerable-population scenarios have
    #    distinct conversation-flow handling.
    identity = identity_context.get(
        "identity", IDENTITY_ADULT_SELF)
    if identity not in trial_context.get(
        "identity_posture", [IDENTITY_ADULT_SELF],
    ):
        # The trial does not enroll this identity class.
        # Route appropriately rather than improvising.
        _emit_event("RecruitmentEvent.IdentityMismatch", {
            "session_id":         session_id,
            "trial_id":           trial_id,
            "identity":           identity,
            "trial_identity_set": trial_context["identity_posture"],
        })
        return {
            "routing":     "IDENTITY_MISMATCH",
            "identity":    identity,
            "session_id":  session_id,
            "trial_id":    trial_id,
            "turn_index":  turn_index,
        }

    # 6. Conversation-state load (or initialize on first turn).
    conversation_state = _safe_get(
        CONVERSATION_STATE_TABLE, {"session_id": session_id})
    if conversation_state is None:
        conversation_state = {
            "session_id":          session_id,
            "trial_id":            trial_id,
            "amendment_version":   trial_context["content_version"],
            "started_at":          _now_iso(),
            "turn_count":          0,
            "funnel_stage":        FUNNEL_STAGE_ENTERED,
            "disclosures_shown":   [],
            "prescreen_state":     "NOT_STARTED",
            "language":            language,
            "channel":             channel,
            "referral_source":     referral_source,
            "identity":            identity,
        }
        record_funnel_stage(
            session_id=session_id,
            trial_id=trial_id,
            stage=FUNNEL_STAGE_ENTERED,
            metadata={"channel": channel,
                      "referral_source": referral_source})
    else:
        conversation_state = _from_decimal(conversation_state)
        # Mid-conversation amendment-version drift triggers
        # a re-presentation flow. The demo logs the drift
        # and continues; production pauses the conversation
        # and surfaces the IRB-approved re-presentation
        # message before resuming.
        if conversation_state["amendment_version"] != \
                trial_context["content_version"]:
            _emit_event(
                "RecruitmentEvent.AmendmentDriftDetected", {
                    "session_id":              session_id,
                    "trial_id":                trial_id,
                    "session_amendment_ver":   conversation_state[
                        "amendment_version"],
                    "current_amendment_ver":   trial_context[
                        "content_version"],
                    "detected_at":             _now_iso(),
                })

    conversation_state["turn_count"] = (
        conversation_state.get("turn_count", 0) + 1)
    conversation_state["last_seen_at"] = _now_iso()

    # 7. Persist the updated state.
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    table.put_item(Item=_to_decimal(conversation_state))

    return {
        "routing":              "AGENT",
        "session_id":           session_id,
        "trial_id":             trial_id,
        "trial_context":        trial_context,
        "trial_state":          trial_state,
        "identity":             identity,
        "conversation_state":   conversation_state,
        "user_message":         user_message,
        "turn_index":           turn_index,
    }

def handle_emergency_routing(turn_result: dict) -> dict:
    """Render the IRB-approved emergency message and pause
    the recruitment conversation."""
    response_text = (
        "I'm not able to help with this kind of urgent "
        "concern. If this is a medical emergency, please "
        "call 911 right now or go to your nearest emergency "
        "room. If you or someone you know is in crisis, "
        "you can also call or text 988 to reach the Suicide "
        "and Crisis Lifeline. If you'd like, after you've "
        "taken care of the immediate concern, you can come "
        "back here and continue learning about the trial.")

    persist_recruitment_decision(
        session_id=turn_result["session_id"],
        trial_id=turn_result["trial_id"],
        disposition=DISPOSITION_EMERGENCY_ROUTED,
        prescreen_summary=None,
        coordinator_handoff=None,
        notes=("Emergency signal detected; "
                "conversation paused per IRB-approved "
                "emergency-routing language."))

    _put_metric("EmergencyRouted", 1, {
        "trial_id": turn_result["trial_id"],
    })
    return {
        "response_text": response_text,
        "routing":       "EMERGENCY",
    }

def handle_out_of_scope_routing(turn_result: dict) -> dict:
    """Render an IRB-approved out-of-scope routing message
    keyed to the matched category."""
    category = turn_result["category"]
    if category == "clinical_advice_about_existing_care":
        response_text = (
            "That's a question about your current care, "
            "and that's outside what I'm here for. I'd "
            "encourage you to talk with your existing care "
            "team about that. If you have questions about "
            "the research trial itself, I'd be glad to help "
            "with those.")
    elif category == "trial_recommendation_request":
        response_text = (
            "I'm not able to recommend whether you should "
            "join the trial. That's a decision for you and "
            "your care team. What I can do is help you "
            "understand what the trial is studying, what "
            "participation involves, and whether you appear "
            "to meet the eligibility criteria. Then a "
            "research coordinator can walk you through the "
            "next steps.")
    elif category == "benefits_quote_request":
        response_text = (
            "Insurance coverage questions are outside what "
            "I'm able to answer. The research coordinator "
            "can walk you through what costs the trial "
            "covers and what your insurance handles. I'd "
            "be glad to connect you with them.")
    elif category == "prescription_request":
        response_text = (
            "I can't help with prescriptions. For medication "
            "questions, please reach out to your existing "
            "care team or your pharmacy. If you have "
            "questions about the trial, I'm here for those.")
    else:
        response_text = (
            "That's outside what I'm here to help with. "
            "If you have questions about the trial, I'd be "
            "glad to help with those.")

    _put_metric("OutOfScopeRouted", 1, {
        "trial_id": turn_result["trial_id"],
        "category": category,
    })
    return {
        "response_text": response_text,
        "routing":       "OUT_OF_SCOPE",
        "category":      category,
    }

def handle_trial_unavailable(turn_result: dict) -> dict:
    """Render an IRB-approved message for closed-or-paused
    trials and route to the appropriate alternative."""
    state = turn_result["trial_state"]["state"]
    if state == TRIAL_STATE_CLOSED:
        response_text = (
            "Thanks for your interest in this trial. "
            "Enrollment for this study has closed. If "
            "you'd like, I can connect you with the "
            "research team for information about other "
            "studies that may be open in this area.")
    elif state == TRIAL_STATE_PAUSED:
        response_text = (
            "Enrollment for this trial is currently paused. "
            "I don't have a date for when it might reopen. "
            "If you'd like, I can connect you with the "
            "research team to learn more.")
    elif state in (
        TRIAL_STATE_AMENDMENT_PENDING,
        TRIAL_STATE_AMENDMENT_REJECTED,
    ):
        response_text = (
            "I'm not able to share details about this trial "
            "right now. The research team is working through "
            "an update to the study. If you'd like, I can "
            "have a research coordinator follow up with you "
            "when there's more to share.")
    else:
        response_text = (
            "I'm not able to help with this trial right now. "
            "Let me connect you with the research team.")

    _put_metric("TrialUnavailableShown", 1, {
        "trial_id":   turn_result["trial_id"],
        "trial_state": state,
    })
    return {
        "response_text": response_text,
        "routing":       "TRIAL_UNAVAILABLE",
        "trial_state":   state,
    }

def handle_identity_mismatch(turn_result: dict) -> dict:
    """The trial does not enroll the identity class of this
    conversation participant. Route appropriately."""
    identity = turn_result["identity"]
    if identity == IDENTITY_PARENT_GUARDIAN:
        response_text = (
            "Thanks for asking on your child's behalf. This "
            "trial enrolls adults only, so this study isn't "
            "a fit. The research team can suggest pediatric "
            "studies that may be open if you'd like.")
    elif identity == IDENTITY_SURROGATE_DECISION_MAKER:
        response_text = (
            "Thanks for reaching out. This trial isn't set "
            "up for surrogate-decision-maker enrollment. "
            "The research team can talk with you about "
            "studies that are.")
    else:
        response_text = (
            "Based on what you've shared, this trial may "
            "not be the right fit. The research team can "
            "talk with you about other studies that might be.")

    _put_metric("IdentityMismatch", 1, {
        "trial_id": turn_result["trial_id"],
        "identity": identity,
    })
    return {
        "response_text": response_text,
        "routing":       "IDENTITY_MISMATCH",
        "identity":      identity,
    }

# Tool surface declared to the orchestration model.
RECRUITMENT_TOOL_SCHEMA = [
    {
        "name": "trial_context_retrieve",
        "description": (
            "Retrieve a section of the IRB-approved trial "
            "context (protocol summary, visit schedule, "
            "study procedures, sponsor information) by "
            "section name. Always cite the returned content."),
        "input_schema": {
            "type": "object",
            "properties": {
                "section": {"type": "string"},
            },
            "required": ["section"],
        },
    },
    {
        "name": "recruitment_faq_retrieve",
        "description": (
            "Retrieve IRB-approved recruitment-FAQ entries "
            "matching the patient's question. Returns "
            "verbatim IRB-approved answer text. Always cite."),
        "input_schema": {
            "type": "object",
            "properties": {
                "question": {"type": "string"},
            },
            "required": ["question"],
        },
    },
    {
        "name": "eligibility_question_surface",
        "description": (
            "Surface an eligibility criterion question to "
            "the patient. The deterministic rule engine "
            "owns the criterion-evaluation logic; the model "
            "captures the patient's response and routes it "
            "to the rule engine via "
            "eligibility_response_capture."),
        "input_schema": {
            "type": "object",
            "properties": {
                "criterion_id": {"type": "string"},
            },
            "required": ["criterion_id"],
        },
    },
    {
        "name": "eligibility_response_capture",
        "description": (
            "Capture the patient's structured response to a "
            "previously-surfaced eligibility-criterion "
            "question and route it to the deterministic "
            "rule evaluator. The evaluator returns MET, "
            "NOT_MET, INDETERMINATE, or "
            "REQUIRES_COORDINATOR_REVIEW."),
        "input_schema": {
            "type": "object",
            "properties": {
                "criterion_id":           {"type": "string"},
                "patient_response_value": {"type": "string"},
                "patient_reported_unit":  {"type": "string"},
            },
            "required": ["criterion_id", "patient_response_value"],
        },
    },
    {
        "name": "prescreen_save_progress",
        "description": (
            "Persist accumulated prescreen state for the "
            "session. Called periodically and at the end "
            "of the prescreen flow."),
        "input_schema": {
            "type": "object",
            "properties": {
                "captured_responses": {"type": "object"},
                "current_disposition": {"type": "string"},
            },
            "required": ["captured_responses",
                          "current_disposition"],
        },
    },
    {
        "name": "coordinator_handoff_request",
        "description": (
            "Request a coordinator handoff once the patient "
            "is at LIKELY_ELIGIBLE_PENDING or "
            "UNCERTAIN_PENDING and remains interested. "
            "Captures preferred follow-up channel and time."),
        "input_schema": {
            "type": "object",
            "properties": {
                "preferred_channel":      {"type": "string"},
                "preferred_time_window":  {"type": "string"},
                "patient_questions_open": {
                    "type": "array",
                    "items": {"type": "string"}},
            },
            "required": ["preferred_channel"],
        },
    },
    {
        "name": "representativeness_capture",
        "description": (
            "Capture, with patient consent, demographic "
            "self-report per the trial's IRB-approved data-"
            "collection plan and the sponsor's diversity "
            "action plan where applicable. Patient may "
            "decline any field without losing access to "
            "the recruitment conversation."),
        "input_schema": {
            "type": "object",
            "properties": {
                "language":           {"type": "string"},
                "race_ethnicity":     {"type": "string"},
                "sex_gender":         {"type": "string"},
                "age_cohort":         {"type": "string"},
                "geography_zip3":     {"type": "string"},
                "insurance_category": {"type": "string"},
            },
        },
    },
    {
        "name": "request_coordinator_immediate",
        "description": (
            "Patient requests to speak with a coordinator "
            "without continuing the prescreen. Capture and "
            "route immediately."),
        "input_schema": {
            "type": "object",
            "properties": {
                "reason": {"type": "string"},
            },
        },
    },
]

def build_system_prompt(trial_context: dict,
                          identity: str,
                          conversation_state: dict) -> str:
    """Build the per-conversation system prompt. Encodes
    the IRB-citation discipline, the scope discipline, and
    the disclosure-required-on-first-turn discipline."""
    lines = [
        f"You are the recruitment assistant for clinical "
        f"trial {trial_context['protocol_identifier']} "
        f"(IRB protocol "
        f"{trial_context['irb_protocol_number']}).",
        "",
        "RULES YOU MUST FOLLOW:",
        "1. Every trial-specific statement must be grounded "
        "   in an IRB-approved retrieval result. If you do "
        "   not have an IRB-approved citation for a claim, "
        "   you must not make the claim. You may say 'I "
        "   don't have that information; let me connect "
        "   you with a research coordinator.'",
        "2. You do NOT recommend whether the patient should "
        "   join the trial. You explain, screen, and route. "
        "   Decisions are the patient's, in partnership "
        "   with their care team.",
        "3. You do NOT diagnose, prescribe, advise on the "
        "   patient's existing care, or quote insurance "
        "   coverage. Route those topics to the appropriate "
        "   alternative pathway via the tool surface.",
        "4. You are a chat tool, not a person. You are not "
        "   the research coordinator. You cannot enroll the "
        "   patient. The patient can stop at any time and "
        "   can ask for the research coordinator at any "
        "   time.",
        "5. For eligibility questions you surface the "
        "   criterion via eligibility_question_surface and "
        "   capture the patient response via "
        "   eligibility_response_capture. The deterministic "
        "   rule engine owns the criterion-evaluation logic. "
        "   You do not interpret eligibility criteria.",
        "6. For factual questions about the trial you use "
        "   recruitment_faq_retrieve or trial_context_retrieve. "
        "   The retrieval surface is the only source of "
        "   trial-specific recruitment language.",
        "7. If the patient asks to be connected to a "
        "   coordinator at any time, you call "
        "   request_coordinator_immediate and confirm the "
        "   handoff arrangement.",
        "8. If you detect that the patient is in distress, "
        "   acknowledge with calibrated language and offer "
        "   to connect them with the coordinator or with "
        "   appropriate support resources. You are not a "
        "   counselor.",
        "",
        f"IDENTITY POSTURE: {identity}",
        f"TRIAL THERAPEUTIC AREA: "
        f"{trial_context['therapeutic_area']}",
        f"AMENDMENT VERSION: "
        f"{trial_context['content_version']}",
    ]

    is_first_turn = (conversation_state.get(
        "turn_count", 0) <= 1)
    if is_first_turn:
        lines.append("")
        lines.append(
            "FIRST-TURN REQUIREMENT: surface the IRB-"
            "approved disclosures before any trial-specific "
            "content:")
        for d in REQUIRED_DISCLOSURES_FIRST_TURN:
            lines.append(f"  - {d}")

    return "\n".join(lines)

def run_agent_turn(turn_result: dict) -> dict:
    """
    Run the agent's tool-use loop for the current turn.
    Returns the structured response dict including the
    raw model text, the tool-call trace, and the
    citations referenced.
    """
    session_id          = turn_result["session_id"]
    trial_id            = turn_result["trial_id"]
    user_message        = turn_result["user_message"]
    trial_context       = turn_result["trial_context"]
    identity            = turn_result["identity"]
    conversation_state  = turn_result["conversation_state"]

    system_prompt = build_system_prompt(
        trial_context, identity, conversation_state)

    # Conversation history (a real implementation reloads
    # the prior turns; the demo is single-turn for clarity).
    messages = [{
        "role":    "user",
        "content": [{"type": "text", "text": user_message}],
    }]

    tool_trace: list = []
    citations:  list = []
    final_text  = ""

    for iteration in range(MAX_AGENT_ITERATIONS):
        # In production this is the real bedrock-runtime
        # invoke_model call with tools and tool_choice
        # parameters. The mock returns scripted responses.
        invoke_body = json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 1500,
            "system":     system_prompt,
            "messages":   messages,
            "tools":      RECRUITMENT_TOOL_SCHEMA,
        })
        response = bedrock_runtime.invoke_model(
            modelId=ORCHESTRATION_MODEL_ID,
            body=invoke_body)
        body_text = response["body"].read().decode("utf-8")
        body = json.loads(body_text)

        stop_reason = body.get("stop_reason", "end_turn")
        content_blocks = body.get("content", [])

        if stop_reason == "tool_use":
            # Execute each tool call and append the result
            # back to the conversation for the next model
            # turn.
            assistant_blocks = []
            tool_results = []
            for block in content_blocks:
                if block.get("type") == "tool_use":
                    tool_name = block["name"]
                    tool_args = block.get("input", {})
                    tool_use_id = block.get(
                        "id", uuid.uuid4().hex)
                    started_at = _now()
                    tool_result = dispatch_tool(
                        tool_name=tool_name,
                        tool_args=tool_args,
                        session_id=session_id,
                        trial_id=trial_id,
                        trial_context=trial_context,
                        conversation_state=conversation_state)
                    elapsed_ms = int(
                        (_now() - started_at).total_seconds()
                        * 1000)
                    _audit_tool_call(
                        session_id=session_id,
                        trial_id=trial_id,
                        tool=tool_name,
                        arguments=tool_args,
                        result_summary=_summarize_tool_result(
                            tool_result),
                        latency_ms=elapsed_ms,
                        outcome=tool_result.get(
                            "outcome", "OK"))
                    tool_trace.append({
                        "tool":       tool_name,
                        "args":       tool_args,
                        "result":     tool_result,
                        "elapsed_ms": elapsed_ms,
                    })
                    if tool_result.get("citations"):
                        citations.extend(
                            tool_result["citations"])
                    assistant_blocks.append(block)
                    tool_results.append({
                        "type":         "tool_result",
                        "tool_use_id":  tool_use_id,
                        "content":      json.dumps(
                            _from_decimal(tool_result)),
                    })
                elif block.get("type") == "text":
                    assistant_blocks.append(block)
            messages.append({"role":    "assistant",
                              "content": assistant_blocks})
            messages.append({"role":    "user",
                              "content": tool_results})
            continue

        # End-of-turn. Collect the final text and any
        # citation references.
        for block in content_blocks:
            if block.get("type") == "text":
                final_text += block["text"]
                citations.extend(block.get("citations", []))
        break
    else:
        logger.warning(
            "Agent loop exhausted MAX_AGENT_ITERATIONS=%s "
            "for session=%s trial=%s",
            MAX_AGENT_ITERATIONS, session_id, trial_id)

    return {
        "response_text":   final_text,
        "tool_trace":      tool_trace,
        "citations":       citations,
        "messages":        messages,
        "iterations_used": iteration + 1,
    }

def _summarize_tool_result(tool_result: dict) -> dict:
    """Strip large payloads before tool-call ledger storage."""
    summary = {}
    for key, value in tool_result.items():
        if key in ("citations", "outcome",
                    "disposition", "evaluation_outcome",
                    "matched_count"):
            summary[key] = value
        elif isinstance(value, (str, int, float, bool, Decimal)):
            summary[key] = value
        else:
            summary[key] = "[complex_value_omitted]"
    return summary

def dispatch_tool(*, tool_name, tool_args, session_id,
                   trial_id, trial_context,
                   conversation_state) -> dict:
    """Tool-call dispatcher. Each branch wraps the
    corresponding tool-implementation function, which in
    production is its own Lambda."""
    if tool_name == "trial_context_retrieve":
        return tool_trial_context_retrieve(
            trial_id=trial_id,
            section=tool_args["section"])
    if tool_name == "recruitment_faq_retrieve":
        return tool_recruitment_faq_retrieve(
            trial_id=trial_id,
            question=tool_args["question"])
    if tool_name == "eligibility_question_surface":
        return tool_eligibility_question_surface(
            trial_id=trial_id,
            criterion_id=tool_args["criterion_id"])
    if tool_name == "eligibility_response_capture":
        return tool_eligibility_response_capture(
            session_id=session_id,
            trial_id=trial_id,
            criterion_id=tool_args["criterion_id"],
            patient_response_value=tool_args[
                "patient_response_value"],
            patient_reported_unit=tool_args.get(
                "patient_reported_unit"))
    if tool_name == "prescreen_save_progress":
        return tool_prescreen_save_progress(
            session_id=session_id,
            trial_id=trial_id,
            captured_responses=tool_args["captured_responses"],
            current_disposition=tool_args[
                "current_disposition"])
    if tool_name == "coordinator_handoff_request":
        return tool_coordinator_handoff_request(
            session_id=session_id,
            trial_id=trial_id,
            preferred_channel=tool_args["preferred_channel"],
            preferred_time_window=tool_args.get(
                "preferred_time_window"),
            patient_questions_open=tool_args.get(
                "patient_questions_open", []))
    if tool_name == "representativeness_capture":
        return tool_representativeness_capture(
            session_id=session_id,
            trial_id=trial_id,
            captured=tool_args)
    if tool_name == "request_coordinator_immediate":
        return tool_request_coordinator_immediate(
            session_id=session_id,
            trial_id=trial_id,
            reason=tool_args.get("reason", ""))
    return {"outcome": "ERROR",
            "error":   f"Unknown tool {tool_name}"}

def tool_trial_context_retrieve(*, trial_id, section) -> dict:
    """Retrieve a section from the IRB-approved trial
    context. Returns the IRB-approved text with citation
    metadata. Production deployments return retrieval
    results from the Bedrock Knowledge Base bound to the
    per-trial corpus."""
    context = get_trial_context(trial_id)
    if not context:
        return {"outcome": "TRIAL_NOT_FOUND",
                "trial_id": trial_id}

    section_map = {
        "protocol_summary":      "protocol_summary_irb_approved",
        "visit_schedule":        "visit_schedule_irb_approved",
        "study_procedures":      "study_procedures_irb_approved",
        "sponsor_information":   "sponsor_information_irb_approved",
        "contact_information":   "contact_information_irb_approved",
    }
    field = section_map.get(section)
    if field is None:
        return {"outcome": "SECTION_NOT_FOUND",
                "section": section}

    return {
        "outcome":  "OK",
        "section":  section,
        "text":     context.get(field, ""),
        "citations": [{
            "trial_id":          trial_id,
            "section":           section,
            "irb_approval_record":
                context.get("irb_approval_record", {}),
            "content_version":
                context.get("content_version"),
        }],
    }

def tool_recruitment_faq_retrieve(*, trial_id, question) -> dict:
    """Retrieve IRB-approved FAQ entries matching the
    question. The retrieval surface is the only source of
    trial-specific recruitment language for FAQ topics."""
    response = bedrock_agent_runtime.retrieve(
        knowledgeBaseId=FAQ_KB_ID,
        retrievalQuery={"text": question},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "filter": {
                    "equals": {
                        "key":   "trial_id",
                        "value": trial_id,
                    },
                },
            },
        })

    matches = response.get("retrievalResults", [])
    # Defense-in-depth: production deployments rely on the
    # index-time metadata filter (vectorSearchConfiguration
    # above) for per-trial isolation correctness. This
    # post-hoc check guards against misconfigured filters
    # or index-rebuild races; it is not the primary
    # isolation mechanism.
    matches = [
        m for m in matches
        if m.get("metadata", {}).get("trial_id") == trial_id
        and m.get("score", 0.0) >= float(RETRIEVAL_SCORE_FLOOR)
    ]

    if not matches:
        return {
            "outcome": "NO_IRB_APPROVED_ANSWER",
            "question": question,
            "guidance": (
                "No IRB-approved answer exists for this "
                "question. Route the question to the "
                "research coordinator rather than improvising."),
        }

    return {
        "outcome":  "OK",
        "matched_count": len(matches),
        "matches": [{
            "faq_id":  m["metadata"]["faq_id"],
            "text":    m["content"]["text"],
            "score":   m["score"],
        } for m in matches],
        "citations": [{
            "trial_id":         trial_id,
            "faq_id":           m["metadata"]["faq_id"],
            "irb_review_record":
                m["metadata"].get("irb_review_record", {}),
            "score":            m["score"],
        } for m in matches],
    }

def tool_eligibility_question_surface(*, trial_id,
                                        criterion_id) -> dict:
    """Surface an eligibility-criterion question. Returns
    the IRB-approved description text plus a structured
    response-shape hint that the LLM uses to formulate a
    natural-language follow-up."""
    record = _safe_get(ELIGIBILITY_RULE_TABLE,
                        {"trial_id":     trial_id,
                         "criterion_id": criterion_id})
    if not record:
        return {"outcome": "CRITERION_NOT_FOUND",
                "criterion_id": criterion_id}

    record = _from_decimal(record)
    return {
        "outcome":               "OK",
        "criterion_id":          criterion_id,
        "description":           record["description_irb_approved"],
        "category":              record["category"],
        "inclusion":             record.get("inclusion", True),
        "rule_definition":       record["rule_definition"],
        "citations": [{
            "trial_id":          trial_id,
            "criterion_id":      criterion_id,
            "irb_review_record":
                record.get("irb_review_record", {}),
        }],
    }

def tool_eligibility_response_capture(*, session_id,
                                        trial_id,
                                        criterion_id,
                                        patient_response_value,
                                        patient_reported_unit
                                        ) -> dict:
    """Capture the patient's response to a criterion and
    route it to the deterministic rule evaluator."""
    record = _safe_get(ELIGIBILITY_RULE_TABLE,
                        {"trial_id":     trial_id,
                         "criterion_id": criterion_id})
    if not record:
        return {"outcome": "CRITERION_NOT_FOUND",
                "criterion_id": criterion_id}
    record = _from_decimal(record)

    evaluation = evaluate_eligibility_criterion(
        rule_definition=record["rule_definition"],
        category=record["category"],
        inclusion=record.get("inclusion", True),
        patient_response_value=patient_response_value,
        patient_reported_unit=patient_reported_unit)

    # Persist the captured response onto the prescreen
    # record. The conversation state's prescreen_responses
    # dict is the running accumulator.
    state = _safe_get(CONVERSATION_STATE_TABLE,
                       {"session_id": session_id})
    if state:
        state = _from_decimal(state)
    else:
        state = {"session_id": session_id, "trial_id": trial_id}

    responses = state.get("prescreen_responses", {})
    # Record prescreen-started on the first captured criterion.
    if not responses:
        record_funnel_stage(
            session_id=session_id,
            trial_id=trial_id,
            stage=FUNNEL_STAGE_PRESCREEN_STARTED,
            metadata={"first_criterion": criterion_id})
    responses[criterion_id] = {
        "value":            patient_response_value,
        "unit":             patient_reported_unit,
        "evaluation":       evaluation["outcome"],
        "rule_version":     record.get("rule_version", "v1"),
        "captured_at":      _now_iso(),
    }
    state["prescreen_responses"] = responses
    state["prescreen_state"] = "IN_PROGRESS"
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    table.put_item(Item=_to_decimal(state))

    return {
        "outcome":            "OK",
        "criterion_id":       criterion_id,
        "evaluation_outcome": evaluation["outcome"],
        "evaluation_reason":  evaluation.get("reason"),
        "patient_reported":   True,
        "coordinator_review_required":
            evaluation["outcome"] in (
                EVAL_OUTCOME_REQUIRES_COORDINATOR,
                EVAL_OUTCOME_INDETERMINATE),
        "citations": [{
            "trial_id":          trial_id,
            "criterion_id":      criterion_id,
            "rule_version":      record.get("rule_version", "v1"),
            "irb_review_record":
                record.get("irb_review_record", {}),
        }],
    }

def evaluate_eligibility_criterion(*,
                                     rule_definition,
                                     category,
                                     inclusion,
                                     patient_response_value,
                                     patient_reported_unit
                                     ) -> dict:
    """
    Deterministic per-criterion evaluator. The rule
    definition is a structured dict; the evaluator applies
    the rule and returns one of MET, NOT_MET, INDETERMINATE,
    or REQUIRES_COORDINATOR_REVIEW.

    Each rule is owned by named clinical leadership with
    version history and IRB-review evidence. The evaluator
    here covers a small subset of patterns to keep the
    demo readable. Production has dozens of rule types
    covering numeric ranges, lab-value windows, medication
    histories, comorbidity patterns, demographic constraints,
    and time-window arithmetic.
    """
    rule_type = rule_definition.get("type")

    if rule_type == "boolean_response":
        # Patient self-reports yes/no for the criterion.
        normalized = patient_response_value.strip().lower()
        if normalized in ("yes", "y", "true", "1"):
            patient_value_bool = True
        elif normalized in ("no", "n", "false", "0"):
            patient_value_bool = False
        else:
            return {"outcome": EVAL_OUTCOME_INDETERMINATE,
                    "reason":  "Could not parse boolean"}
        expected = rule_definition.get(
            "expected_value", True)
        is_match = (patient_value_bool == expected)
        outcome = (
            EVAL_OUTCOME_MET if (is_match == inclusion)
            else EVAL_OUTCOME_NOT_MET)
        return {"outcome": outcome,
                "reason":  None}

    if rule_type == "age_range":
        try:
            age = int(patient_response_value)
        except (ValueError, TypeError):
            return {"outcome": EVAL_OUTCOME_INDETERMINATE,
                    "reason":  "Could not parse age"}
        min_age = rule_definition.get("min_age", 0)
        max_age = rule_definition.get("max_age", 200)
        in_range = (min_age <= age <= max_age)
        outcome = (
            EVAL_OUTCOME_MET if (in_range == inclusion)
            else EVAL_OUTCOME_NOT_MET)
        return {"outcome": outcome,
                "reason":  None}

    if rule_type == "numeric_range_with_unit":
        # Numeric range with required unit. Includes unit-
        # conversion logic for the common cases (HbA1c %
        # vs mmol/mol, weight kg vs lb).
        try:
            value = float(patient_response_value)
        except (ValueError, TypeError):
            return {"outcome": EVAL_OUTCOME_INDETERMINATE,
                    "reason":  "Could not parse numeric value"}

        target_unit = rule_definition.get("unit")
        if patient_reported_unit and \
                patient_reported_unit != target_unit:
            value = _convert_unit(
                value=value,
                from_unit=patient_reported_unit,
                to_unit=target_unit)
            if value is None:
                return {
                    "outcome": EVAL_OUTCOME_REQUIRES_COORDINATOR,
                    "reason":  (
                        f"Unit conversion from "
                        f"{patient_reported_unit} to "
                        f"{target_unit} not supported")}

        min_v = rule_definition.get("min", float("-inf"))
        max_v = rule_definition.get("max", float("inf"))
        in_range = (min_v <= value <= max_v)
        outcome = (
            EVAL_OUTCOME_MET if (in_range == inclusion)
            else EVAL_OUTCOME_NOT_MET)
        return {"outcome": outcome,
                "reason":  None}

    if rule_type == "categorical_set":
        # Patient response must be in a named set.
        allowed = set(rule_definition.get("allowed_values", []))
        is_match = (
            patient_response_value.strip().lower() in
            {a.lower() for a in allowed})
        outcome = (
            EVAL_OUTCOME_MET if (is_match == inclusion)
            else EVAL_OUTCOME_NOT_MET)
        return {"outcome": outcome,
                "reason":  None}

    if rule_type == "time_since":
        # "How long since X" in days, months, or years.
        # Patient response interpreted as the duration.
        try:
            value = int(patient_response_value)
        except (ValueError, TypeError):
            return {"outcome": EVAL_OUTCOME_INDETERMINATE,
                    "reason":  "Could not parse duration"}
        max_duration_days = rule_definition.get(
            "max_duration_days", float("inf"))
        unit = (patient_reported_unit or "days").lower()
        duration_days = value * {
            "day":   1, "days":   1,
            "week":  7, "weeks":  7,
            "month": 30, "months": 30,
            "year":  365, "years":  365,
        }.get(unit, 1)
        in_window = (duration_days <= max_duration_days)
        outcome = (
            EVAL_OUTCOME_MET if (in_window == inclusion)
            else EVAL_OUTCOME_NOT_MET)
        return {"outcome": outcome,
                "reason":  None}

    if rule_type == "clinical_judgment":
        # The criterion requires clinical judgment (severity,
        # prognosis, comorbidity context). The assistant
        # captures the patient's report; the coordinator
        # makes the determination.
        return {"outcome": EVAL_OUTCOME_REQUIRES_COORDINATOR,
                "reason":  "Clinical judgment required"}

    if rule_type == "verification_only":
        # Patient self-reports; coordinator verifies against
        # the chart or external records.
        return {"outcome": EVAL_OUTCOME_REQUIRES_COORDINATOR,
                "reason":  "Coordinator verification required"}

    return {"outcome": EVAL_OUTCOME_INDETERMINATE,
            "reason":  f"Unknown rule type: {rule_type}"}

def _convert_unit(*, value, from_unit, to_unit) -> Optional[float]:
    """Small unit-conversion table. Production uses a
    clinical-leadership-owned conversion library reviewed
    for the specific clinical contexts the trials use."""
    if from_unit == to_unit:
        return value
    conversions = {
        # HbA1c
        ("%", "mmol/mol"):    lambda v: 10.929 * (v - 2.15),
        ("mmol/mol", "%"):    lambda v: (v / 10.929) + 2.15,
        # Weight
        ("kg", "lb"):         lambda v: v * 2.20462,
        ("lb", "kg"):         lambda v: v / 2.20462,
        # Height
        ("cm", "in"):         lambda v: v / 2.54,
        ("in", "cm"):         lambda v: v * 2.54,
    }
    fn = conversions.get((from_unit, to_unit))
    return fn(value) if fn else None

def tool_prescreen_save_progress(*, session_id, trial_id,
                                   captured_responses,
                                   current_disposition) -> dict:
    """Persist the running prescreen state. Called
    periodically and at the end of the prescreen flow."""
    table = dynamodb.Table(CONVERSATION_STATE_TABLE)
    state = _safe_get(CONVERSATION_STATE_TABLE,
                       {"session_id": session_id})
    if state:
        state = _from_decimal(state)
    else:
        state = {"session_id": session_id,
                 "trial_id":   trial_id}
    state["prescreen_responses"] = captured_responses
    state["prescreen_state"] = current_disposition
    state["last_progress_at"] = _now_iso()
    table.put_item(Item=_to_decimal(state))

    if current_disposition == DISPOSITION_DISQUALIFIED:
        _put_metric("PrescreenDisqualified", 1, {
            "trial_id": trial_id})
        record_funnel_stage(
            session_id=session_id,
            trial_id=trial_id,
            stage=FUNNEL_STAGE_PRESCREEN_COMPLETED,
            metadata={"disposition": current_disposition})
    elif current_disposition == DISPOSITION_LIKELY_ELIGIBLE:
        _put_metric("PrescreenLikelyEligible", 1, {
            "trial_id": trial_id})
        record_funnel_stage(
            session_id=session_id,
            trial_id=trial_id,
            stage=FUNNEL_STAGE_PRESCREEN_COMPLETED,
            metadata={"disposition": current_disposition})
    elif current_disposition == DISPOSITION_UNCERTAIN_PENDING:
        _put_metric("PrescreenUncertainPending", 1, {
            "trial_id": trial_id})
        record_funnel_stage(
            session_id=session_id,
            trial_id=trial_id,
            stage=FUNNEL_STAGE_PRESCREEN_COMPLETED,
            metadata={"disposition": current_disposition})

    return {"outcome": "OK",
            "current_disposition": current_disposition}

def compute_prescreen_disposition(prescreen_responses: dict) -> str:
    """Aggregate per-criterion evaluation outcomes into a
    final disposition. The deterministic engine owns the
    disposition vocabulary: if any criterion evaluated to
    NOT_MET, the patient is disqualified. If all evaluated
    criteria are MET, the patient is likely eligible. If
    any criteria are INDETERMINATE or REQUIRES_COORDINATOR,
    the disposition is uncertain pending coordinator review.
    If no criteria have been evaluated yet, default to
    LIKELY_ELIGIBLE (the patient expressed interest and was
    not disqualified by any evaluated criterion)."""
    if not prescreen_responses:
        return DISPOSITION_LIKELY_ELIGIBLE

    outcomes = [
        r.get("evaluation", EVAL_OUTCOME_INDETERMINATE)
        for r in prescreen_responses.values()
    ]
    if any(o == EVAL_OUTCOME_NOT_MET for o in outcomes):
        return DISPOSITION_DISQUALIFIED
    if any(o in (EVAL_OUTCOME_INDETERMINATE,
                 EVAL_OUTCOME_REQUIRES_COORDINATOR)
           for o in outcomes):
        return DISPOSITION_UNCERTAIN_PENDING
    return DISPOSITION_LIKELY_ELIGIBLE