Recipe 10.7: Python Implementation Example

pip install boto3

pip install amazon-transcribe

import hashlib
import json
import logging
import uuid
from datetime import datetime, timezone
from decimal import Decimal

import boto3
from botocore.config import Config

# Structured logging. In production, ship JSON-formatted records
# to CloudWatch Logs Insights. The ambient documentation
# pipeline operates on heavily PHI-bearing data: the audio is
# PHI and biometric, the verbatim transcript is PHI, the
# generated note is PHI, the structured-field extractions are
# PHI, and every signature event is a clinical-record
# transaction. Log structural metadata only (session_id, ASR
# confidence band, faithfulness score band, signature event),
# never raw transcripts, never patient demographics, never
# medication or diagnosis values.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from HealthScribe, Bedrock,
# Comprehend Medical, DynamoDB, S3, EventBridge, CloudWatch,
# and Secrets Manager. The streaming-display latency budget is
# tight (a couple of seconds end to end); the post-encounter
# batch budget is looser (a minute or two is acceptable). Cap
# the retries on the streaming path so the in-encounter
# display does not stall; let the batch path retry more
# aggressively because the clinician is no longer waiting in
# real time.
BOTO3_RETRY_CONFIG_STREAMING = Config(
    retries={"max_attempts": 2, "mode": "adaptive"})
BOTO3_RETRY_CONFIG_BATCH = Config(
    retries={"max_attempts": 5, "mode": "adaptive"})

# Module-level clients. Reused across Lambda invocations in
# warm containers so each invocation does not pay the
# connection cost. The demo below uses Mock* classes instead;
# the real clients are never invoked here. Note: streaming
# HealthScribe and streaming Transcribe Medical do not run
# through the boto3 transcribe client; they use the standalone
# amazon-transcribe-streaming-sdk Python package.
REGION = "us-east-1"
dynamodb               = boto3.resource("dynamodb", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
s3_client              = boto3.client("s3", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
transcribe_batch       = boto3.client("transcribe", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
bedrock_runtime        = boto3.client("bedrock-runtime", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
comprehend_medical     = boto3.client("comprehendmedical", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
eventbridge_client     = boto3.client("events", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
cloudwatch_client      = boto3.client("cloudwatch", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
secrets_client         = boto3.client("secretsmanager", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)
stepfunctions_client   = boto3.client("stepfunctions", region_name=REGION,
                                          config=BOTO3_RETRY_CONFIG_BATCH)

# --- Resource Names ---
# Fill these in with your actual resource names. The demo
# prints what it would write rather than failing if the
# resources do not exist; see run_demo() at the bottom.
ENCOUNTER_STATE_TABLE     = "ambient-doc-encounter-state"
TRANSCRIPT_STATE_TABLE    = "ambient-doc-transcript-state"
NOTE_STATE_TABLE          = "ambient-doc-note-state"
AUDIO_BUCKET              = "ambient-doc-audio"
TRANSCRIPT_BUCKET         = "ambient-doc-transcripts"
AUDIT_ARCHIVE_BUCKET      = "ambient-doc-audit-archive"
HEALTHSCRIBE_OUTPUT_BUCKET = "ambient-doc-healthscribe-output"
ENCOUNTER_EVENT_BUS_NAME  = "ambient-doc-events-bus"
CLOUDWATCH_NAMESPACE      = "AmbientDocumentation"
INSTITUTION_ID            = "academic-medical-center-richmond"
INSTITUTIONAL_VOCABULARY  = "ambient-doc-clinical-vocabulary"
INSTITUTIONAL_LANGUAGE_MODEL = "ambient-doc-clinical-lm-v2"
AMBIENT_DOC_GUARDRAIL_ID  = "guardrail-90123"
AMBIENT_DOC_GUARDRAIL_VERSION = "2"
HEALTHSCRIBE_DATA_ACCESS_ROLE_ARN = (
    "arn:aws:iam::000000000000:role/HealthScribeDataAccessRole")
OUTPUT_KMS_KEY_ARN = (
    "arn:aws:kms:us-east-1:000000000000:key/"
    "11111111-2222-3333-4444-555555555555")

# Bedrock configuration. In production, pin to a specific
# model version and inference profile so a model upgrade does
# not silently change note-rendering behavior. The model and
# region combination must be in your AWS BAA scope.
BEDROCK_NOTE_RENDERING_MODEL_ID = (
    "anthropic.claude-3-5-sonnet-20240620-v1:0")
BEDROCK_NOTE_RENDERING_PROFILE_ARN = (
    "arn:aws:bedrock:us-east-1:000000000000:inference-profile/"
    "ambient-doc-note-render-v1")
# A smaller, cheaper model is appropriate for the faithfulness
# check; the check is structurally simpler than the rendering.
BEDROCK_FAITHFULNESS_MODEL_ID = (
    "anthropic.claude-3-haiku-20240307-v1:0")
BEDROCK_FAITHFULNESS_PROFILE_ARN = (
    "arn:aws:bedrock:us-east-1:000000000000:inference-profile/"
    "ambient-doc-faithfulness-v1")
# Structured extraction uses the same Sonnet-class model as
# note rendering, but with a different prompt and a strict
# JSON-schema response format expressed via tool-use.
BEDROCK_EXTRACTION_MODEL_ID = (
    "anthropic.claude-3-5-sonnet-20240620-v1:0")

# Deploy-time guardrail. Any blank resource name is a deploy-
# time bug, not a runtime surprise.
for _name, _value in [
    ("ENCOUNTER_STATE_TABLE",     ENCOUNTER_STATE_TABLE),
    ("TRANSCRIPT_STATE_TABLE",    TRANSCRIPT_STATE_TABLE),
    ("NOTE_STATE_TABLE",          NOTE_STATE_TABLE),
    ("AUDIO_BUCKET",              AUDIO_BUCKET),
    ("TRANSCRIPT_BUCKET",         TRANSCRIPT_BUCKET),
    ("AUDIT_ARCHIVE_BUCKET",      AUDIT_ARCHIVE_BUCKET),
    ("HEALTHSCRIBE_OUTPUT_BUCKET", HEALTHSCRIBE_OUTPUT_BUCKET),
    ("ENCOUNTER_EVENT_BUS_NAME",  ENCOUNTER_EVENT_BUS_NAME),
    ("CLOUDWATCH_NAMESPACE",      CLOUDWATCH_NAMESPACE),
    ("INSTITUTIONAL_VOCABULARY",  INSTITUTIONAL_VOCABULARY),
    ("BEDROCK_NOTE_RENDERING_MODEL_ID",
        BEDROCK_NOTE_RENDERING_MODEL_ID),
    ("BEDROCK_FAITHFULNESS_MODEL_ID",
        BEDROCK_FAITHFULNESS_MODEL_ID),
]:
    assert _value, f"{_name} must be set before deploying."

# --- Versioning ---
# Every encounter record carries the versions of the artifacts
# that influenced it: the streaming ASR model version, the
# batch ASR model version, the note-rendering prompt version,
# the faithfulness-check prompt version, the structured-
# extraction rules version, the per-specialty template
# version. A future audit reconstructs which configuration was
# active when a particular encounter was processed.
STREAMING_ASR_MODEL_VERSION   = "healthscribe-streaming-2026-q1"
BATCH_ASR_MODEL_VERSION       = "healthscribe-batch-2026-q1"
NOTE_RENDERING_PROMPT_VERSION = "note-render-prompt-v2.1"
FAITHFULNESS_PROMPT_VERSION   = "faithfulness-prompt-v1.3"
STRUCTURED_EXTRACTION_VERSION = "structured-extraction-v1.2"
TEMPLATE_LIBRARY_VERSION      = "templates-v3.0"

# --- ASR Confidence and Audio Quality Gates ---
# Average per-word confidence below this floor flags the
# transcript for QA review and surfaces a low-confidence flag
# to the reviewing clinician. The minimum-confidence-word
# threshold is a secondary gate: even a high-average-
# confidence transcript with several critical words at very
# low confidence (often medication names) deserves an explicit
# review highlight.
ASR_MIN_AVG_CONFIDENCE              = Decimal("0.85")
ASR_LOW_CONFIDENCE_WORD_THRESHOLD   = Decimal("0.70")
ASR_QA_REVIEW_LOW_CONF_WORD_LIMIT   = 5

# Audio quality SNR warning threshold. When per-encounter
# average SNR drops below this, the clinician's device gets
# an audio-quality warning so they can investigate the room
# acoustics or microphone placement before the encounter
# proceeds.
AUDIO_QUALITY_WARNING_THRESHOLD_DB  = Decimal("12.0")

# --- Faithfulness Threshold ---
# The faithfulness checker returns a score in [0, 1]; below
# this threshold, the draft is either blocked (severity=block)
# or flagged for the clinician's attention (severity=flag).
# Severity is determined per failure type by the institutional
# faithfulness program.
FAITHFULNESS_PASS_THRESHOLD     = Decimal("0.88")
FAITHFULNESS_BLOCK_THRESHOLD    = Decimal("0.65")

# --- Recording Consent Disclosures ---
# Production looks up the clinic's jurisdiction (for in-person
# ambient documentation, the clinic's location governs) and
# selects the appropriate disclosure. Behavioral-health visits
# use a different disclosure that explicitly mentions
# transcript handling and shorter retention.
CONSENT_DISCLOSURE_ALL_PARTY = (
    "Today's visit will be captured and transcribed for your "
    "medical record. To continue, please confirm by saying "
    "yes when the doctor asks.")
CONSENT_DISCLOSURE_ONE_PARTY = (
    "Today's visit will be captured and transcribed for your "
    "medical record.")
CONSENT_DISCLOSURE_BEHAVIORAL_HEALTH = (
    "Today's visit will be captured and transcribed for your "
    "medical record. The transcript will be retained as part "
    "of your behavioral health treatment record under our "
    "confidentiality policy. Please confirm to continue.")

# All-party-consent states. The list is approximate and
# changes over time as state law evolves; production maintains
# this in a legal-team-reviewed configuration with an explicit
# update cadence.
# TODO (TechWriter): verify the current all-party-consent state
# list against the Reporters Committee for Freedom of the Press
# state-by-state recording-laws guide before deploying.
ALL_PARTY_CONSENT_STATES = {
    "CA", "CT", "FL", "IL", "MD", "MA",
    "MT", "NV", "NH", "PA", "WA",
}

# Visit types that always trigger an explicit per-encounter
# disclosure regardless of the state regime, because the
# content is more sensitive and the patient's expectation of
# conversational privacy is higher.
SENSITIVE_VISIT_TYPES = {
    "behavioral_health",
    "substance_use_treatment",
    "reproductive_health_sensitive",
    "genetic_counseling",
}

# --- Cohort Axes ---
# Stratification axes used in the audit pipeline for equity
# monitoring. The age-band and primary-language are opt-in:
# the patient self-discloses during portal enrollment.
# Inferred demographic labels for protected classes are
# explicitly not used.
COHORT_AXES = ["language", "visit_type", "specialty",
               "audio_quality_band", "age_band", "room_id"]

# --- Per-Specialty Note Templates ---
# Production maintains per-specialty templates as versioned
# clinical-informatics assets. The demo includes a small
# subset to show the pattern.
DEFAULT_TEMPLATES = {
    "primary-care-soap-v3": {
        "id":       "primary-care-soap-v3",
        "specialty": "family_medicine",
        "structure": "SOAP",
        "sections": ["subjective", "objective",
                     "assessment", "plan"],
    },
    "behavioral-health-progress-v2": {
        "id":       "behavioral-health-progress-v2",
        "specialty": "behavioral_health",
        "structure": "behavioral_health_progress",
        "sections": ["presenting_concerns",
                     "mental_status_exam",
                     "risk_assessment", "interventions",
                     "plan"],
    },
    "geriatric-comprehensive-v1": {
        "id":       "geriatric-comprehensive-v1",
        "specialty": "geriatrics",
        "structure": "SOAP_with_function",
        "sections": ["subjective", "functional_status",
                     "objective", "assessment", "plan"],
    },
}

# --- Helper: float -> Decimal ---
# DynamoDB rejects native Python float. Every numeric value on
# its way into DynamoDB has to be a Decimal. This helper
# handles nested dicts and lists.
def _to_decimal(value):
    if isinstance(value, float):
        return Decimal(str(value))
    if isinstance(value, dict):
        return {k: _to_decimal(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_to_decimal(v) for v in value]
    return value

def _hash_value(value):
    """Stable hash for non-PHI audit linkage."""
    if not value:
        return None
    return hashlib.sha256(str(value).encode("utf-8")).hexdigest()

def _now_iso():
    return datetime.now(timezone.utc).isoformat()

def _bucket_audio_quality(quality_metrics):
    """
    Categorize per-encounter audio quality into a small number
    of bands for cohort stratification. Production tunes these
    thresholds against the institution's actual audio-quality
    distribution. For in-room ambient documentation, the room
    acoustics and microphone placement drive most of the
    variation; the band is one of the strongest predictors of
    downstream transcript and note quality.
    """
    if not quality_metrics:
        return "unknown"
    avg_snr = quality_metrics.get("avg_snr_db", 0)
    if avg_snr >= 25:
        return "high"
    if avg_snr >= 15:
        return "medium"
    return "low"

class MockHealthScribeStreaming:
    """
    Stands in for the streaming HealthScribe API used during
    the encounter. In production, the streaming API runs over
    HTTP/2 via the amazon-transcribe Python package and emits
    speaker-attributed transcript segments with clinical-
    content classification labels. The mock plays back canned
    per-segment streaming events so the pipeline can be
    exercised without a real audio source.
    """
    def __init__(self, fixture_segments):
        # fixture_segments is keyed by encounter_id, with each
        # entry a list of segment dicts that simulate what
        # would arrive on the streaming connection.
        self._fixtures = fixture_segments
        self.streamed = []

    def stream_session(self, session_id, encounter_id,
                         expected_speaker_count, language):
        """
        Yields streaming-event dicts from the in-room
        microphone. Production yields these as they arrive;
        the mock yields them in the order specified by the
        fixture, preserving each segment's speaker_role and
        segment_class so the downstream handlers see the same
        event shape they would get from real HealthScribe.
        """
        segments = self._fixtures.get(encounter_id, [])
        for segment in segments:
            event = {
                "session_id":      session_id,
                "speaker_role":    segment["speaker_role"],
                "transcript":      segment["text"],
                "is_partial":      False,
                "timestamp":       segment["timestamp"],
                "segment_class":
                    segment.get("segment_class", "hpi"),
                "diarization_confidence":
                    segment.get("diarization_confidence",
                                Decimal("0.92")),
                "average_word_confidence":
                    segment.get("confidence", Decimal("0.9")),
                "words": [{"word": w,
                           "confidence":
                               segment.get("confidence",
                                            Decimal("0.9"))}
                           for w in segment["text"].split()],
            }
            self.streamed.append(event)
            yield event

    def emit_audio_quality_events(self, session_id,
                                     encounter_id):
        """
        Production emits periodic audio-quality events
        (signal-to-noise ratio, speech-detection rate,
        acoustic-event detections) on the streaming session.
        The mock yields a single fixture quality reading so
        the downstream handler exercises the warning path
        when the room is noisy.
        """
        # The fixture key is per-encounter so different
        # scenarios can simulate different acoustic conditions.
        fixtures = {
            "encounter-2026-05-23-0411": {
                "signal_to_noise_db": Decimal("22.5"),
                "speech_detection_rate": Decimal("0.62"),
                "acoustic_events": [],
            },
            "encounter-2026-05-23-0412": {
                "signal_to_noise_db": Decimal("9.5"),
                "speech_detection_rate": Decimal("0.48"),
                "acoustic_events": [
                    {"type": "hallway_bleed",
                     "timestamp": "00:04:21"},
                    {"type": "door_open",
                     "timestamp": "00:11:08"},
                ],
            },
        }
        yield fixtures.get(encounter_id, {
            "signal_to_noise_db": Decimal("20.0"),
            "speech_detection_rate": Decimal("0.55"),
            "acoustic_events": [],
        })

class MockHealthScribeBatch:
    """
    Stands in for the batch HealthScribe API used after the
    encounter ends. In production this is
    transcribe_batch.start_medical_scribe_job(...) which
    produces a higher-accuracy transcript with full
    diarization plus a structured clinical note draft. The
    mock returns a canned full-encounter transcript and the
    HealthScribe-default SOAP note draft.
    """
    def __init__(self, fixture_transcripts, fixture_note_drafts):
        self._fixture_transcripts = fixture_transcripts
        self._fixture_note_drafts = fixture_note_drafts
        self.jobs_started = []

    def start_medical_scribe_job(self, job_name, encounter_id,
                                    audio_uri, language,
                                    max_speaker_labels,
                                    template_id):
        self.jobs_started.append({
            "job_name":           job_name,
            "encounter_id":       encounter_id,
            "audio_uri":          audio_uri,
            "language":           language,
            "max_speaker_labels": max_speaker_labels,
            "template_id":        template_id,
            "started_at":         _now_iso(),
        })
        # Return a fake job descriptor that "completes
        # immediately" for the demo.
        return {"MedicalScribeJobName": job_name,
                "MedicalScribeJobStatus": "COMPLETED"}

    def retrieve_transcript(self, encounter_id):
        return dict(self._fixture_transcripts.get(
            encounter_id, {}))

    def retrieve_note_draft(self, encounter_id):
        return dict(self._fixture_note_drafts.get(
            encounter_id, {}))

class MockBedrock:
    """
    Stands in for Amazon Bedrock's InvokeModel. Three
    invocation patterns: institutional-template note rendering
    (returns a structured note draft with citations,
    re-rendered from HealthScribe's default SOAP draft into
    the institution's preferred format), faithfulness check
    (returns a 0-to-1 score plus a list of any failed checks),
    and structured extraction (returns a JSON-schema-validated
    extraction object for higher-level fields Comprehend
    Medical does not directly extract).
    """
    def __init__(self, render_responses, faithfulness_responses,
                 extraction_responses):
        self._render_responses = render_responses
        self._faithfulness_responses = faithfulness_responses
        self._extraction_responses = extraction_responses
        self.invocations = []

    def render_institutional_note(self, encounter_id,
                                     transcript,
                                     healthscribe_draft,
                                     template, guardrail_id):
        # Production: bedrock_runtime.invoke_model with the
        # transcript, the HealthScribe SOAP draft, and the
        # institutional template structure in the prompt
        # context. Structured output for Anthropic Claude on
        # Bedrock is enforced through tool-use (the tools
        # field inside the body, with the structured-output
        # schema declared as a tool definition) or through a
        # system prompt that demands JSON. The guardrail is
        # applied at runtime via guardrailIdentifier and
        # guardrailVersion parameters on invoke_model.
        self.invocations.append({
            "type":         "note_rendering",
            "encounter_id": encounter_id,
            "template":     template["id"],
            "guardrail_id": guardrail_id,
        })
        response = self._render_responses.get(encounter_id, {
            "content":   {"sections": {}},
            "citations": [],
            "guardrail_action": "NONE",
        })
        return {"body": json.dumps(response,
                                     default=str)}

    def check_faithfulness(self, encounter_id, rendered_note,
                            transcript, ehr_context):
        # Production: bedrock_runtime.invoke_model with a
        # smaller model, prompted to score the rendered note
        # against the source transcript and EHR context for
        # citation grounding, absence of hallucinated content,
        # and clinical consistency.
        self.invocations.append({
            "type":         "faithfulness_check",
            "encounter_id": encounter_id,
        })
        response = self._faithfulness_responses.get(
            encounter_id, {
                "score":         0.95,
                "failed_checks": [],
                "annotations":   [],
            })
        return {"body": json.dumps(response,
                                     default=str)}

    def extract_higher_level_fields(self, encounter_id,
                                      transcript):
        # Production: bedrock_runtime.invoke_model for entities
        # Comprehend Medical does not directly extract (orders,
        # follow-up actions, patient-reported vitals,
        # patient-reported allergies, referrals).
        self.invocations.append({
            "type":         "structured_extraction",
            "encounter_id": encounter_id,
        })
        response = self._extraction_responses.get(
            encounter_id, {
                "orders_placed":           [],
                "labs_requested":          [],
                "imaging_requested":       [],
                "follow_up_appointments":  [],
                "patient_reported_vitals": [],
                "patient_reported_allergies": [],
                "referrals_placed":        [],
            })
        return {"body": json.dumps(response,
                                     default=str)}

class MockComprehendMedical:
    """
    Stands in for Amazon Comprehend Medical's
    DetectEntitiesV2, InferRxNorm, and InferICD10CM APIs.
    Used to extract coded medication and condition entities
    from the transcript with RxNorm and ICD-10 linking.
    """
    def __init__(self, entity_fixtures):
        self._fixtures = entity_fixtures
        self.invocations = []

    def detect_entities(self, text):
        self.invocations.append({"type": "detect_entities",
                                 "text_len": len(text)})
        return self._fixtures.get("detect_entities",
                                    {"Entities": []})

    def infer_rx_norm(self, text):
        self.invocations.append({"type": "infer_rx_norm",
                                 "text_len": len(text)})
        return self._fixtures.get("infer_rx_norm",
                                    {"Entities": []})

    def infer_icd10cm(self, text):
        self.invocations.append({"type": "infer_icd10cm",
                                 "text_len": len(text)})
        return self._fixtures.get("infer_icd10cm",
                                    {"Entities": []})

class MockEHR:
    """
    Stands in for the EHR's FHIR write surface for clinical
    notes (DocumentReference), structured chart updates
    (MedicationRequest, Condition, Observation, ServiceRequest),
    and patient-portal release. In production this is the EHR
    vendor's FHIR API authenticated through SMART on FHIR or
    backend-services authentication.
    """
    def __init__(self):
        self.documents_written = []
        self.chart_updates = []
        self.portal_releases = []

    def fetch_ehr_context(self, patient_id_hash):
        """
        Production reads the patient's allergies, current
        medications, problem list, recent labs, and recent
        imaging from the EHR to populate note sections that
        are not usually discussed aloud (and to give the
        faithfulness check a second grounding source). The
        mock returns a small synthetic context.
        """
        return {
            "allergies":     ["NKDA"],
            "medications":   [
                {"name": "metformin",  "dose": "1000 mg BID",
                 "rx_norm_code": "860975"},
                {"name": "lisinopril", "dose": "10 mg daily",
                 "rx_norm_code": "314076"},
            ],
            "problems":      [
                {"name": "Type 2 diabetes mellitus",
                 "icd_10_code": "E11.9"},
                {"name": "Essential hypertension",
                 "icd_10_code": "I10"},
            ],
            "recent_labs":   [],
            "recent_imaging": [],
        }

    def write_document_reference(self, patient_id, encounter_id,
                                   document_content, author,
                                   signed_at,
                                   idempotency_key):
        document_id = f"doc-{uuid.uuid4().hex[:10]}"
        self.documents_written.append({
            "document_id":      document_id,
            "patient_id":       patient_id,
            "encounter_id":     encounter_id,
            "document_content": document_content,
            "author":           author,
            "signed_at":        signed_at,
            "idempotency_key":  idempotency_key,
        })
        return {"document_id": document_id}

    def apply_structured_update(self, patient_id, update_kind,
                                  payload):
        self.chart_updates.append({
            "patient_id":   patient_id,
            "update_kind":  update_kind,  # medication, problem, etc.
            "payload":      dict(payload),
            "applied_at":   _now_iso(),
        })

    def release_to_portal(self, patient_id, summary,
                           release_at):
        self.portal_releases.append({
            "patient_id": patient_id,
            "summary":    summary,
            "release_at": release_at,
        })

class MockClinicianReviewClient:
    """
    Stands in for the clinician's review-and-sign web client.
    The demo simulates the clinician opening the review pane,
    confirming or rejecting structured-field extractions,
    making narrative edits, and signing.
    """
    def __init__(self):
        self._review_decisions = []
        self._signature = None
        self._note_edits = None

    def queue_review_decisions(self, decisions):
        self._review_decisions = list(decisions)

    def queue_note_edits(self, edits):
        self._note_edits = edits

    def queue_signature(self, signature):
        self._signature = dict(signature)

    def collect_review_decisions(self):
        decisions = list(self._review_decisions)
        self._review_decisions = []
        return decisions

    def collect_note_edits(self):
        edits = self._note_edits
        self._note_edits = None
        return edits

    def get_signature(self):
        signature = dict(self._signature or {
            "type":      "electronic",
            "method":    "password_plus_mfa",
            "timestamp": _now_iso(),
        })
        self._signature = None
        return signature

class MockEncounterState:
    def __init__(self):
        self._items = {}

    def get(self, session_id):
        return dict(self._items.get(session_id, {}))

    def put(self, session_id, item):
        self._items[session_id] = dict(item)

    def update(self, session_id, updates):
        existing = self._items.setdefault(
            session_id, {"session_id": session_id})
        existing.update(updates)

class MockTranscriptState:
    def __init__(self):
        self._items = {}

    def get(self, session_id):
        return dict(self._items.get(session_id, {}))

    def put(self, session_id, item):
        self._items[session_id] = dict(item)

    def update(self, session_id, updates):
        existing = self._items.setdefault(
            session_id, {"session_id": session_id})
        existing.update(updates)

    def append_streaming_segment(self, session_id, segment):
        existing = self._items.setdefault(
            session_id, {"session_id": session_id,
                          "streaming_segments": []})
        existing.setdefault("streaming_segments", []).append(
            dict(segment))

    def get_streaming_segments(self, session_id):
        return list(self._items.get(session_id, {})
                     .get("streaming_segments", []))

class MockNoteState:
    def __init__(self):
        self._items = {}

    def get(self, session_id):
        return dict(self._items.get(session_id, {}))

    def put(self, session_id, item):
        self._items[session_id] = dict(item)

    def update(self, session_id, updates):
        existing = self._items.setdefault(
            session_id, {"session_id": session_id})
        existing.update(updates)

class MockS3:
    """
    Stands in for S3 audio storage, transcript storage,
    HealthScribe output, and audit archive. Holds objects in
    memory keyed by (bucket, key). Production uses customer-
    managed KMS keys for encryption and lifecycle policies for
    retention.
    """
    def __init__(self):
        self._objects = {}

    def put_object(self, bucket, key, body, metadata=None):
        self._objects[(bucket, key)] = {
            "body":      body,
            "metadata":  dict(metadata or {}),
            "stored_at": _now_iso(),
        }
        return {"bucket": bucket, "key": key,
                "uri":    f"s3://{bucket}/{key}"}

    def get_object(self, bucket, key):
        return dict(self._objects.get((bucket, key), {}))

    def list(self, bucket=None):
        return [{"bucket": b, "key": k,
                 "metadata": v["metadata"]}
                for (b, k), v in self._objects.items()
                if bucket is None or b == bucket]

class MockEventBus:
    """
    Stands in for Amazon EventBridge. Lifecycle events flow
    here for cross-system fan-out: session_started,
    encounter_transcribed, note_generated, note_signed,
    encounter_audited.
    """
    def __init__(self):
        self.events = []

    def put_events(self, entries):
        for entry in entries:
            self.events.append(dict(entry))

class MockCloudWatch:
    """
    Stands in for CloudWatch metric emission. In production
    the metrics flow into dashboards and alarms.
    """
    def __init__(self):
        self.metrics = []

    def put_metric(self, namespace, metric_name, value,
                    unit="Count", dimensions=None):
        self.metrics.append({
            "namespace":   namespace,
            "metric_name": metric_name,
            "value":       value,
            "unit":        unit,
            "dimensions":  dimensions or {},
            "timestamp":   _now_iso(),
        })

# Module-level singletons for the demo. In production each of
# these is its own AWS resource accessed via boto3.
encounter_state        = MockEncounterState()
transcript_state       = MockTranscriptState()
note_state             = MockNoteState()
ehr                    = MockEHR()
clinician_client       = MockClinicianReviewClient()
event_bus              = MockEventBus()
cloudwatch             = MockCloudWatch()
s3_store               = MockS3()
# healthscribe_streaming, healthscribe_batch_mock,
# bedrock_mock, and comprehend_mock are wired up in run_demo()
# with fixture data tailored to each scenario.
healthscribe_streaming = None
healthscribe_batch_mock = None
bedrock_mock           = None
comprehend_mock        = None

# Clinician voiceprint registry. Production stores enrolled
# clinician voiceprints (with consent, with biometric-data
# governance) and references them at session start so the
# diarization can label clinician segments directly. The mock
# stores opaque IDs.
CLINICIAN_VOICEPRINT_REGISTRY = {
    "clinician-patel":   "voiceprint-patel-abc123",
    "clinician-okonkwo": "voiceprint-okonkwo-def456",
}

def audit_log(event):
    """
    Sanitized audit print so you can see the sequence of
    decisions without leaking the underlying values.
    Production routes events to CloudWatch Logs Insights with
    structured JSON; ship to a SIEM if available.
    """
    safe_event = {k: v for k, v in event.items()
                  if k not in {"verbatim_transcript",
                                "rendered_note_text",
                                "patient_demographics",
                                "structured_decisions_raw"}}
    if "verbatim_transcript" in event:
        safe_event["verbatim_transcript_length"] = len(
            event["verbatim_transcript"] or "")
    if "rendered_note_text" in event:
        safe_event["rendered_note_text_length"] = len(
            event["rendered_note_text"] or "")
    logger.info("AUDIT %s", json.dumps(safe_event,
                                          default=str))

def determine_consent_regime(clinic_jurisdiction, visit_type):
    """
    For in-person ambient documentation, the clinic's location
    governs the recording-consent regime. Behavioral-health
    and other sensitive visit types use an explicit-
    acknowledgment disclosure regardless of the state regime.
    """
    if visit_type in SENSITIVE_VISIT_TYPES:
        return "behavioral_health_explicit"
    if clinic_jurisdiction in ALL_PARTY_CONSENT_STATES:
        return "all_party_consent"
    return "one_party_consent"

def select_disclosure_text(consent_regime):
    """Return the disclosure text for the regime."""
    if consent_regime == "behavioral_health_explicit":
        return CONSENT_DISCLOSURE_BEHAVIORAL_HEALTH
    if consent_regime == "all_party_consent":
        return CONSENT_DISCLOSURE_ALL_PARTY
    return CONSENT_DISCLOSURE_ONE_PARTY

def lookup_feature_status(clinician_id, visit_type):
    """
    Per-clinician and per-visit-type feature status lookup.
    Some clinicians opt out entirely; some institutions
    exclude certain visit types (behavioral-health is the
    most common exclusion).
    """
    # Production reads from a per-institution policy table.
    # The demo defaults all encounters to enabled.
    return {"enabled": True, "reason": "default_enabled"}

def configure_audio_capture(room_id, session_id, device_type,
                              expected_speaker_count):
    """
    Configure the in-room audio capture path. Production
    reads the per-room device metadata (which device is
    deployed, what its capabilities are) and sets up the
    streaming session. The demo collapses this to a config
    dict that downstream stages reference.
    """
    return {
        "room_id":                room_id,
        "session_id":             session_id,
        "device_type":            device_type,
        # Possible device_type values:
        #   clinician_tablet, wall_mounted_array,
        #   ceiling_mounted_array, lavalier_with_room_array.
        "encoding":               "pcm",
        "sample_rate":            16000,
        "expected_speaker_count": expected_speaker_count,
        "beamforming_enabled":
            device_type != "clinician_tablet",
        "noise_suppression":      "moderate",
        "vad_enabled":            True,
        "audio_stream_path":
            f"audio/{session_id}/encounter.pcm",
    }

def check_voiceprint_enrollment(clinician_id):
    """
    Look up whether the clinician has an enrolled voiceprint.
    Voiceprint enrollment is biometric-data governance
    territory (Illinois BIPA, Texas, Washington each have
    specific consent and disclosure rules); production gates
    enrollment behind explicit clinician consent and an
    institutional review.
    """
    return clinician_id in CLINICIAN_VOICEPRINT_REGISTRY


def _lookup_voiceprint_consent_version(clinician_id):
    """
    Return the version identifier of the biometric-data consent
    the clinician signed at voiceprint enrollment. Returns None
    if the clinician has not enrolled.
    """
    if clinician_id in CLINICIAN_VOICEPRINT_REGISTRY:
        return "bipa-consent-v2.1"
    return None


def _lookup_biometric_jurisdiction(room_id):
    """
    Determine the biometric-data jurisdiction from the room's
    physical location. The clinic's state determines which
    biometric-data law applies (BIPA in Illinois, CUBI in
    Texas, RCW 19.375 in Washington, or none).
    """
    # In production, room_id maps to a clinic site which maps
    # to a state. The demo returns a fixed jurisdiction.
    return "IL"

def encounter_start(encounter_id, patient_id, clinician_id,
                     clinician_specialty, clinic_jurisdiction,
                     visit_type, room_id,
                     bystanders=None,
                     device_type="wall_mounted_array",
                     language="en-US",
                     consent_acknowledged=True):
    """
    Start an ambient-documentation session in response to an
    encounter-start event. Captures consent, identifies
    bystanders, and bootstraps the encounter state.

    Args:
        encounter_id: Encounter identifier from the EHR.
        patient_id: Patient identifier (will be hashed in
            persistent state).
        clinician_id: Authenticated clinician identifier.
        clinician_specialty: Specialty key used for template
            selection.
        clinic_jurisdiction: Two-letter state code for the
            clinic's location (governs recording-consent
            regime for in-person encounters).
        visit_type: e.g. "primary_care", "behavioral_health",
            "geriatric_followup".
        room_id: Per-room identifier so the audit can stratify
            by room (room acoustics drive per-cohort accuracy).
        bystanders: List of bystander descriptors (e.g.
            ["family_member", "caregiver"]).
        device_type: In-room capture device type.
        language: Encounter language. Defaults to en-US.
        consent_acknowledged: For the demo, whether the
            patient acknowledged the disclosure.

    Returns:
        Session-context dict consumed by downstream stages, or
        a sentinel indicating the patient declined or the
        feature was disabled.
    """
    bystanders = list(bystanders or [])

    # Step 1A: determine the recording-consent regime.
    consent_regime = determine_consent_regime(
        clinic_jurisdiction=clinic_jurisdiction,
        visit_type=visit_type)
    disclosure_text = select_disclosure_text(consent_regime)

    # Step 1B: determine whether ambient documentation is
    # enabled for this clinician + visit-type combination.
    feature_status = lookup_feature_status(
        clinician_id=clinician_id, visit_type=visit_type)
    if not feature_status["enabled"]:
        audit_log({
            "event_type":   "AMBIENT_FEATURE_DISABLED",
            "encounter_id": encounter_id,
            "reason":       feature_status["reason"],
            "timestamp":    _now_iso(),
        })
        return {"feature_enabled": False,
                "reason":           feature_status["reason"]}

    # Step 1C: capture consent disclosure. For all-party-
    # consent jurisdictions and sensitive visit types, gate
    # continuation on explicit acknowledgment. Production
    # plays the disclosure through the in-room device's
    # speaker; the demo records the disclosure text.
    requires_acknowledgment = consent_regime in (
        "all_party_consent", "behavioral_health_explicit")
    if requires_acknowledgment and not consent_acknowledged:
        audit_log({
            "event_type":     "AMBIENT_DECLINED_BY_PATIENT",
            "encounter_id":   encounter_id,
            "consent_regime": consent_regime,
            "timestamp":      _now_iso(),
        })
        return {"feature_enabled": False,
                "reason":           "consent_declined",
                "consent_regime":   consent_regime}

    # Step 1D: bootstrap the ambient session.
    session_id = "amb-" + uuid.uuid4().hex[:16]
    started_at = _now_iso()

    # The clinician declares who is in the room. The
    # diarization layer uses the expected speaker count;
    # the consent record captures who is being captured.
    expected_speaker_count = 1 + len(bystanders)  # patient + others

    audio_capture_config = configure_audio_capture(
        room_id=room_id,
        session_id=session_id,
        device_type=device_type,
        expected_speaker_count=expected_speaker_count)

    voiceprint_enrolled = check_voiceprint_enrollment(
        clinician_id)

    encounter_state.put(session_id, _to_decimal({
        "session_id":              session_id,
        "encounter_id":            encounter_id,
        "patient_id_hash":         _hash_value(patient_id),
        "clinician_id":            clinician_id,
        "clinician_specialty":     clinician_specialty,
        "room_id":                 room_id,
        "visit_type":              visit_type,
        "language":                language,
        "consent_regime":          consent_regime,
        "bystanders":              bystanders,
        "expected_speaker_count":  expected_speaker_count,
        "feature_status":          "enabled",
        "started_at":              started_at,
        "device_type":             device_type,
        "audio_capture_config":    audio_capture_config,
        "audio_archive_ref":       None,
        "canonical_transcript_ref": None,
        "healthscribe_note_draft_ref": None,
        "clinician_voiceprint_enrolled": voiceprint_enrolled,
        "clinician_voiceprint_consent_version":
            _lookup_voiceprint_consent_version(clinician_id),
        "clinician_jurisdiction_for_biometric_compliance":
            _lookup_biometric_jurisdiction(room_id),
        "avg_audio_quality_snr":   Decimal("0.0"),
        "avg_streaming_asr_confidence": Decimal("0.0"),
        "avg_batch_asr_confidence":     Decimal("0.0"),
        "diarization_disagreement_count": 0,
        "streaming_asr_model_version":   STREAMING_ASR_MODEL_VERSION,
        "batch_asr_model_version":       BATCH_ASR_MODEL_VERSION,
        "note_rendering_prompt_version":
            NOTE_RENDERING_PROMPT_VERSION,
        "faithfulness_prompt_version":
            FAITHFULNESS_PROMPT_VERSION,
        "structured_extraction_version":
            STRUCTURED_EXTRACTION_VERSION,
        "template_library_version":
            TEMPLATE_LIBRARY_VERSION,
    }))

    # Step 1E: emit lifecycle event.
    event_bus.put_events([{
        "Source":       "ambient_documentation",
        "DetailType":   "session_started",
        "EventBusName": ENCOUNTER_EVENT_BUS_NAME,
        "Detail": json.dumps({
            "session_id":      session_id,
            "encounter_id":    encounter_id,
            "visit_type":      visit_type,
            "language":        language,
            "consent_regime":  consent_regime,
            "bystander_count": len(bystanders),
            "device_type":     device_type,
            "room_id":         room_id,
            "timestamp":       started_at,
        }),
    }])

    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "SessionsStarted", 1, "Count",
        dimensions={"visit_type":   visit_type,
                    "language":     language,
                    "specialty":    clinician_specialty,
                    "device_type":  device_type})

    audit_log({
        "event_type":              "AMBIENT_SESSION_OPENED",
        "session_id":              session_id,
        "encounter_id":            encounter_id,
        "visit_type":              visit_type,
        "consent_regime":          consent_regime,
        "language":                language,
        "device_type":             device_type,
        "expected_speaker_count":  expected_speaker_count,
        "voiceprint_enrolled":     voiceprint_enrolled,
        "timestamp":               started_at,
    })

    return {
        "feature_enabled":      True,
        "session_id":           session_id,
        "encounter_id":         encounter_id,
        "patient_id":           patient_id,
        "clinician_id":         clinician_id,
        "clinician_specialty":  clinician_specialty,
        "room_id":              room_id,
        "visit_type":           visit_type,
        "language":             language,
        "consent_regime":       consent_regime,
        "bystanders":           bystanders,
        "expected_speaker_count": expected_speaker_count,
        "device_type":          device_type,
        "audio_capture_config": audio_capture_config,
        "voiceprint_enrolled":  voiceprint_enrolled,
        "started_at":           started_at,
        "disclosure_text":      disclosure_text,
    }

def push_audio_quality_warning(session_id, quality):
    """
    Production pushes an audio-quality warning to the
    clinician's device or to a discreet display. The mock
    just logs the warning. Common sources of quality drops:
    HVAC noise, microphone too far from active speaker,
    adjacent-room sound bleed, the clinician walking out of
    the device's beamforming sweet spot.
    """
    audit_log({
        "event_type":           "AUDIO_QUALITY_WARNING",
        "session_id":           session_id,
        "signal_to_noise_db":
            float(quality.get("signal_to_noise_db", 0)),
        "speech_detection_rate":
            float(quality.get("speech_detection_rate", 0)),
        "acoustic_event_count":
            len(quality.get("acoustic_events", [])),
        "timestamp":            _now_iso(),
    })

def stream_audio_to_healthscribe(session_context):
    """
    Run the streaming HealthScribe session for the encounter.
    HealthScribe handles ASR, diarization, role assignment
    (CLINICIAN, PATIENT, FAMILY, OTHER), and clinical-content
    classification together; the demo's mock simulates the
    same per-segment event flow.
    """
    session_id = session_context["session_id"]
    encounter_id = session_context["encounter_id"]
    audio_capture_config = session_context["audio_capture_config"]
    language = session_context["language"]
    state = encounter_state.get(session_id)

    # Step 2A: configure the streaming session. Production
    # passes max_speaker_labels (so HealthScribe knows the
    # expected speaker count from the clinician's bystander
    # declaration), the institutional vocabulary, and the
    # clinician voiceprint hint when enrollment is available.
    #
    # In production:
    #   client = TranscribeStreamingClient(region=REGION)
    #   stream = await client.start_stream_transcription(
    #       language_code=language,
    #       media_sample_rate_hz=audio_capture_config["sample_rate"],
    #       media_encoding=audio_capture_config["encoding"],
    #       vocabulary_name=INSTITUTIONAL_VOCABULARY,
    #       language_model_name=INSTITUTIONAL_LANGUAGE_MODEL,
    #       show_speaker_label=True,
    #       max_speaker_labels=audio_capture_config[
    #           "expected_speaker_count"],
    #       enable_partial_results_stabilization=True)
    #
    # HealthScribe-specific streaming exposes role labels
    # (CLINICIAN, PATIENT, FAMILY, OTHER) and clinical
    # classification (CHIEF_COMPLAINT, HPI, ROS, EXAM, etc.).
    # The mock yields canned segments with these labels
    # already assigned.

    streaming_confidence_sum = Decimal("0.0")
    streaming_segment_count = 0
    diarization_confidence_sum = Decimal("0.0")

    # Step 2B: process each segment as it arrives.
    for event in healthscribe_streaming.stream_session(
            session_id=session_id,
            encounter_id=encounter_id,
            expected_speaker_count=
                audio_capture_config["expected_speaker_count"],
            language=language):

        handle_streaming_segment(
            session_id=session_id,
            event=event,
            language=language,
            specialty=session_context["clinician_specialty"])

        streaming_confidence_sum += Decimal(str(
            event["average_word_confidence"]))
        diarization_confidence_sum += Decimal(str(
            event["diarization_confidence"]))
        streaming_segment_count += 1

    # Step 2C: monitor per-encounter audio quality.
    avg_snr = Decimal("0.0")
    for quality in healthscribe_streaming.emit_audio_quality_events(
            session_id=session_id, encounter_id=encounter_id):
        avg_snr = quality.get("signal_to_noise_db",
                              Decimal("0.0"))
        cloudwatch.put_metric(
            CLOUDWATCH_NAMESPACE, "AudioQualitySNR",
            float(avg_snr), "None",
            dimensions={
                "room_id":     state.get("room_id", "unknown"),
                "device_type":
                    state.get("device_type", "unknown"),
                "visit_type":
                    session_context["visit_type"],
            })
        if avg_snr < AUDIO_QUALITY_WARNING_THRESHOLD_DB:
            push_audio_quality_warning(session_id, quality)
        # Persist acoustic events to the audit pipeline. A
        # door-open event might explain a brief diarization
        # confusion later in the encounter; a hallway-bleed
        # event might explain low-confidence segments that
        # should be excluded from the note.
        for acoustic_event in quality.get("acoustic_events", []):
            audit_log({
                "event_type":           "ACOUSTIC_EVENT",
                "session_id":           session_id,
                "acoustic_event_type":
                    acoustic_event.get("type"),
                "acoustic_event_timestamp":
                    acoustic_event.get("timestamp"),
                "timestamp":            _now_iso(),
            })

    avg_streaming_confidence = (
        (streaming_confidence_sum / streaming_segment_count)
        if streaming_segment_count else Decimal("0.0"))
    avg_diarization_confidence = (
        (diarization_confidence_sum / streaming_segment_count)
        if streaming_segment_count else Decimal("0.0"))

    encounter_state.update(session_id, _to_decimal({
        "avg_audio_quality_snr":         avg_snr,
        "avg_streaming_asr_confidence":  avg_streaming_confidence,
        "avg_diarization_confidence":
            avg_diarization_confidence,
        "streaming_segment_count":       streaming_segment_count,
    }))

    audit_log({
        "event_type":              "STREAMING_SESSION_COMPLETED",
        "session_id":              session_id,
        "segments_processed":      streaming_segment_count,
        "avg_streaming_confidence":
            float(avg_streaming_confidence),
        "avg_diarization_confidence":
            float(avg_diarization_confidence),
        "avg_audio_quality_snr":   float(avg_snr),
        "timestamp":               _now_iso(),
    })

    return {
        "session_id":                session_id,
        "segments_processed":        streaming_segment_count,
        "avg_streaming_confidence":  avg_streaming_confidence,
        "avg_diarization_confidence":
            avg_diarization_confidence,
        "avg_audio_quality_snr":     avg_snr,
    }

def handle_streaming_segment(session_id, event, language,
                                specialty):
    """
    Persist the streaming segment metadata (verbatim text
    goes to the transcript-archive S3 bucket rather than
    DynamoDB to avoid creating a parallel PHI store outside
    the standard audit governance), update the live display,
    and emit per-segment quality metrics.
    """
    transcript_state.append_streaming_segment(
        session_id=session_id,
        segment={
            "speaker_role":   event["speaker_role"],
            "text":           event["transcript"],
            "is_final":       not event["is_partial"],
            "words":          event["words"],
            "timestamp":      event["timestamp"],
            "segment_class":  event["segment_class"],
            "average_word_confidence":
                event["average_word_confidence"],
            "diarization_confidence":
                event["diarization_confidence"],
        })

    # Production pushes the segment to the clinician's live
    # display via an API Gateway WebSocket or server-sent
    # event. The demo just records the metric.
    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "StreamingASRConfidence",
        float(event["average_word_confidence"]), "None",
        dimensions={
            "speaker_role": event["speaker_role"],
            "language":     language,
            "specialty":    specialty,
        })
    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "DiarizationConfidence",
        float(event["diarization_confidence"]), "None",
        dimensions={
            "speaker_role": event["speaker_role"],
            "language":     language,
            "specialty":    specialty,
        })

def select_template(visit_type, specialty):
    """
    Production maintains per-specialty per-visit-type
    templates as versioned clinical-informatics assets. The
    demo uses a small fixed mapping.
    """
    if visit_type == "behavioral_health":
        return DEFAULT_TEMPLATES["behavioral-health-progress-v2"]
    if specialty == "geriatrics":
        return DEFAULT_TEMPLATES["geriatric-comprehensive-v1"]
    return DEFAULT_TEMPLATES["primary-care-soap-v3"]

def run_batch_healthscribe(session_context,
                              audio_archive_ref):
    """
    Trigger the batch HealthScribe job over the full audio.
    HealthScribe batch produces a higher-accuracy transcript
    plus a structured clinical note draft organized by
    section (Subjective, Objective, Assessment, Plan by
    default). The mock returns canned fixtures.
    """
    session_id = session_context["session_id"]
    encounter_id = session_context["encounter_id"]
    state = encounter_state.get(session_id)
    template = select_template(
        visit_type=session_context["visit_type"],
        specialty=session_context["clinician_specialty"])

    # In production:
    #   transcribe_batch.start_medical_scribe_job(
    #       MedicalScribeJobName=f"{session_id}-batch",
    #       Media={"MediaFileUri": audio_archive_ref},
    #       OutputBucketName=HEALTHSCRIBE_OUTPUT_BUCKET,
    #       OutputEncryptionKMSKeyId=OUTPUT_KMS_KEY_ARN,
    #       DataAccessRoleArn=
    #           HEALTHSCRIBE_DATA_ACCESS_ROLE_ARN,
    #       Settings={
    #           "ShowSpeakerLabels": True,
    #           "MaxSpeakerLabels":
    #               state["expected_speaker_count"],
    #           "ChannelIdentification": False,
    #           "VocabularyName": INSTITUTIONAL_VOCABULARY,
    #           "ClinicalNoteGenerationSettings": {
    #               "NoteTemplate": template["id"],
    #           },
    #       })
    # Then poll get_medical_scribe_job until COMPLETED. The
    # mock collapses this to a single call that returns
    # immediately.
    job_name = f"{session_id}-batch-{uuid.uuid4().hex[:6]}"
    job = healthscribe_batch_mock.start_medical_scribe_job(
        job_name=job_name,
        encounter_id=encounter_id,
        audio_uri=audio_archive_ref,
        language=session_context["language"],
        max_speaker_labels=state.get("expected_speaker_count",
                                     2),
        template_id=template["id"])

    canonical_transcript = (
        healthscribe_batch_mock.retrieve_transcript(
            encounter_id=encounter_id))
    healthscribe_note_draft = (
        healthscribe_batch_mock.retrieve_note_draft(
            encounter_id=encounter_id))

    # Step 3D: persist the canonical transcript and the
    # HealthScribe-default note draft to S3 with KMS
    # encryption. Production additionally writes references
    # to the encounter-state and transcript-state tables.
    transcript_object = s3_store.put_object(
        bucket=TRANSCRIPT_BUCKET,
        key=f"{session_id}/canonical_transcript.json",
        body=json.dumps(canonical_transcript,
                          default=str).encode("utf-8"),
        metadata={"session_id":   session_id,
                   "encounter_id": encounter_id})

    note_draft_object = s3_store.put_object(
        bucket=HEALTHSCRIBE_OUTPUT_BUCKET,
        key=f"{session_id}/healthscribe_note_draft.json",
        body=json.dumps(healthscribe_note_draft,
                          default=str).encode("utf-8"),
        metadata={"session_id":   session_id,
                   "encounter_id": encounter_id})

    # Compute average batch confidence for the cohort metrics.
    batch_confidence_sum = Decimal("0.0")
    batch_segment_count = 0
    for seg in canonical_transcript.get("segments", []):
        batch_confidence_sum += Decimal(str(
            seg.get("confidence", 0.9)))
        batch_segment_count += 1
    avg_batch_confidence = (
        (batch_confidence_sum / batch_segment_count)
        if batch_segment_count else Decimal("0.0"))

    encounter_state.update(session_id, _to_decimal({
        "canonical_transcript_ref":    transcript_object["uri"],
        "healthscribe_note_draft_ref":
            note_draft_object["uri"],
        "batch_completed_at":          _now_iso(),
        "avg_batch_asr_confidence":    avg_batch_confidence,
    }))

    transcript_state.update(session_id, _to_decimal({
        "canonical_transcript_ref": transcript_object["uri"],
        "batch_segment_count":       batch_segment_count,
    }))

    event_bus.put_events([{
        "Source":       "ambient_documentation",
        "DetailType":   "encounter_transcribed",
        "EventBusName": ENCOUNTER_EVENT_BUS_NAME,
        "Detail": json.dumps({
            "session_id":     session_id,
            "encounter_id":   encounter_id,
            "avg_batch_confidence":
                float(avg_batch_confidence),
            "segment_count":  batch_segment_count,
        }),
    }])

    audit_log({
        "event_type":              "BATCH_HEALTHSCRIBE_COMPLETED",
        "session_id":              session_id,
        "job_name":                job["MedicalScribeJobName"],
        "segment_count":           batch_segment_count,
        "avg_batch_confidence":
            float(avg_batch_confidence),
        "canonical_transcript_ref":
            transcript_object["uri"],
        "timestamp":               _now_iso(),
    })

    return {
        "canonical_transcript":     canonical_transcript,
        "healthscribe_note_draft":  healthscribe_note_draft,
        "canonical_transcript_ref": transcript_object["uri"],
    }

def lookup_clinician_style(clinician_id):
    """
    Per-clinician style preferences (terse SOAP vs. narrative
    HPI prose, specific phrasing preferences). Production
    learns these from the clinician's prior signed notes (with
    consent) or captures them through explicit configuration.
    The demo returns an empty preference set.
    """
    return {"prefer_narrative_hpi": False,
            "prefer_terse_assessment": False,
            "specific_phrasings": []}

def determine_faithfulness_severity(failed_checks, score):
    """
    Map faithfulness-check results to a severity level.
    Production's mapping is policy-owned by the clinical-
    quality officer; the demo collapses to a simple threshold
    plus a check-type lookup. Severe failures (claim_without_
    citation, contradiction_with_transcript) block the draft
    from being shown at all; minor failures flag for clinician
    attention.
    """
    if score < FAITHFULNESS_BLOCK_THRESHOLD:
        return "block"

    severe_check_types = {
        "claim_without_citation",
        "contradiction_with_transcript",
        "added_clinical_recommendation",
        "fabricated_exam_finding",
        "fabricated_medication",
    }
    for check in failed_checks or []:
        if check.get("type") in severe_check_types:
            return "block"
        if check.get("severity") == "severe":
            return "block"

    if score < FAITHFULNESS_PASS_THRESHOLD:
        return "flag"
    return "pass"

def run_faithfulness_check(session_context, rendered_note,
                             canonical_transcript, ehr_context):
    """
    Verify that every claim in the rendered note has a
    transcript citation or an EHR-source citation, score the
    rendered content for citation grounding, and detect
    contradictions or out-of-scope additions.

    Production runs three layers:
      Layer 1: citation grounding verification, schema
        validation, exam-finding-fabrication detection
        (blocks on failure).
      Layer 2: LLM-judge faithfulness scoring, clinical-
        rule-based contradiction detection (warns or blocks
        depending on behavioral-health profile).
      Layer 3: offline sampling review stratified by
        specialty, room, audio-quality band, and language.

    The demo collapses to a single LLM call that
    approximates the combined scoring.
    """
    session_id = session_context["session_id"]
    encounter_id = session_context["encounter_id"]

    response = bedrock_mock.check_faithfulness(
        encounter_id=encounter_id,
        rendered_note=rendered_note,
        transcript=canonical_transcript,
        ehr_context=ehr_context)

    body = json.loads(response["body"])
    score = Decimal(str(body.get("score", 0.0)))
    failed_checks = body.get("failed_checks", [])
    annotations = body.get("annotations", [])

    severity = determine_faithfulness_severity(
        failed_checks=failed_checks, score=score)

    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "FaithfulnessScore",
        float(score), "None",
        dimensions={
            "specialty":  session_context["clinician_specialty"],
            "language":   session_context["language"],
            "visit_type": session_context["visit_type"],
        })
    if severity != "pass":
        cloudwatch.put_metric(
            CLOUDWATCH_NAMESPACE, "FaithfulnessFailures",
            1, "Count",
            dimensions={
                "severity":   severity,
                "specialty":  session_context["clinician_specialty"],
                "language":   session_context["language"],
            })

    audit_log({
        "event_type":           "FAITHFULNESS_CHECK_RUN",
        "session_id":           session_id,
        "score":                float(score),
        "severity":             severity,
        "failed_check_count":   len(failed_checks),
        "timestamp":            _now_iso(),
    })

    return {
        "score":         score,
        "severity":      severity,
        "failed_checks": failed_checks,
        "annotations":   annotations,
    }

def render_institutional_note(session_context,
                                 canonical_transcript,
                                 healthscribe_note_draft):
    """
    Render the institutional-template note from the
    HealthScribe draft, the canonical transcript, and the
    EHR context. Run the faithfulness check before persisting
    the draft. If the faithfulness check blocks the draft,
    fall back to manual documentation and surface the failure
    to the clinical-quality team via the audit pipeline.
    """
    session_id = session_context["session_id"]
    encounter_id = session_context["encounter_id"]
    state = encounter_state.get(session_id)

    # Step 4A: load the per-specialty template.
    template = select_template(
        visit_type=session_context["visit_type"],
        specialty=session_context["clinician_specialty"])

    # Step 4B: assemble EHR context. Production reads from
    # the EHR's FHIR API; the mock returns a synthetic
    # context dict.
    ehr_context = ehr.fetch_ehr_context(
        patient_id_hash=state.get("patient_id_hash"))

    # Step 4D: invoke Bedrock to render the note. Production
    # calls bedrock_runtime.invoke_model with the transcript
    # as grounding source, the HealthScribe SOAP draft as a
    # starting point, the EHR context as additional grounding,
    # the institutional template structure, and the per-
    # clinician style preferences. The Guardrails policy is
    # applied at runtime.
    note_response = bedrock_mock.render_institutional_note(
        encounter_id=encounter_id,
        transcript=canonical_transcript,
        healthscribe_draft=healthscribe_note_draft,
        template=template,
        guardrail_id=AMBIENT_DOC_GUARDRAIL_ID)

    note_body = json.loads(note_response["body"])
    rendered_content = note_body.get("content", {})
    citations = note_body.get("citations", [])
    guardrail_action = note_body.get("guardrail_action", "NONE")

    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "NoteRenderingInvocations",
        1, "Count",
        dimensions={
            "specialty": session_context["clinician_specialty"],
            "template":  template["id"],
        })

    # Check for Guardrail intervention. Production routes
    # this to the clinical-quality team for review; the
    # clinician falls back to manual documentation.
    if guardrail_action == "INTERVENED":
        audit_log({
            "event_type":      "GUARDRAIL_BLOCKED_NOTE",
            "session_id":      session_id,
            "guardrail_id":    AMBIENT_DOC_GUARDRAIL_ID,
            "timestamp":       _now_iso(),
        })
        note_state.put(session_id, _to_decimal({
            "session_id":      session_id,
            "draft_available": False,
            "block_reason":    "guardrail_intervention",
            "fallback":        "manual_documentation",
            "generated_at":    _now_iso(),
        }))
        return {"draft_available": False,
                "reason":          "guardrail_intervention",
                "fallback":        "manual_documentation"}

    # Step 4E: faithfulness check. Verify that every claim in
    # the rendered note has a transcript or EHR citation and
    # that the cited content actually supports the claim.
    faithfulness_result = run_faithfulness_check(
        session_context=session_context,
        rendered_note=rendered_content,
        canonical_transcript=canonical_transcript,
        ehr_context=ehr_context)

    if faithfulness_result["severity"] == "block":
        # The draft is too unreliable to show; the clinician
        # falls back to manual documentation. The audit
        # captures the failure for clinical-quality review.
        audit_log({
            "event_type":           "NOTE_DRAFT_BLOCKED",
            "session_id":           session_id,
            "faithfulness_score":
                float(faithfulness_result["score"]),
            "failed_checks":
                faithfulness_result["failed_checks"],
            "timestamp":            _now_iso(),
        })
        note_state.put(session_id, _to_decimal({
            "session_id":           session_id,
            "draft_available":      False,
            "block_reason":         "faithfulness_block",
            "faithfulness_score":
                faithfulness_result["score"],
            "faithfulness_failed_checks":
                faithfulness_result["failed_checks"],
            "fallback":             "manual_documentation",
            "generated_at":         _now_iso(),
            "model_version":
                BEDROCK_NOTE_RENDERING_MODEL_ID,
            "prompt_version":
                NOTE_RENDERING_PROMPT_VERSION,
        }))
        return {"draft_available": False,
                "reason":          "faithfulness_block",
                "fallback":        "manual_documentation"}

    # Step 4F: persist the rendered draft with citations and
    # faithfulness annotations.
    rendered_object = s3_store.put_object(
        bucket=TRANSCRIPT_BUCKET,
        key=f"{session_id}/rendered_note.json",
        body=json.dumps({
            "rendered_content": rendered_content,
            "citations":        citations,
            "faithfulness_annotations":
                faithfulness_result["annotations"],
        }, default=str).encode("utf-8"),
        metadata={"session_id": session_id})

    note_state.put(session_id, _to_decimal({
        "session_id":              session_id,
        "draft_available":         True,
        "draft_note":              rendered_content,
        "rendered_note_archive_ref":
            rendered_object["uri"],
        "citations":               citations,
        "template_id":             template["id"],
        "specialty":
            session_context["clinician_specialty"],
        "faithfulness_score":
            faithfulness_result["score"],
        "faithfulness_severity":
            faithfulness_result["severity"],
        "faithfulness_failed_checks":
            faithfulness_result["failed_checks"],
        "faithfulness_annotations":
            faithfulness_result["annotations"],
        "generated_at":            _now_iso(),
        "model_version":
            BEDROCK_NOTE_RENDERING_MODEL_ID,
        "prompt_version":
            NOTE_RENDERING_PROMPT_VERSION,
    }))

    event_bus.put_events([{
        "Source":       "ambient_documentation",
        "DetailType":   "note_generated",
        "EventBusName": ENCOUNTER_EVENT_BUS_NAME,
        "Detail": json.dumps({
            "session_id":   session_id,
            "encounter_id": encounter_id,
            "specialty":
                session_context["clinician_specialty"],
            "faithfulness_score":
                float(faithfulness_result["score"]),
            "faithfulness_severity":
                faithfulness_result["severity"],
        }),
    }])

    audit_log({
        "event_type":          "NOTE_RENDERED",
        "session_id":          session_id,
        "specialty":
            session_context["clinician_specialty"],
        "template_id":         template["id"],
        "faithfulness_score":
            float(faithfulness_result["score"]),
        "faithfulness_severity":
            faithfulness_result["severity"],
        "timestamp":           _now_iso(),
    })

    return {"draft_available":     True,
            "draft_id":            session_id,
            "faithfulness_score":
                faithfulness_result["score"],
            "faithfulness_severity":
                faithfulness_result["severity"]}

def lookup_speaker_role_for_offset(transcript, offset_seconds):
    """
    Find which speaker said the words at a given offset. Used
    for context-aware structured-field extraction (a medication
    mentioned by the patient describing their history is
    processed differently from one the clinician verbalizes
    as part of the plan).
    """
    for segment in transcript.get("segments", []):
        # The fixture timestamps are HH:MM:SS strings;
        # normalize to seconds for the comparison.
        ts = segment.get("timestamp", "00:00:00")
        try:
            parts = [int(p) for p in ts.split(":")]
            seg_seconds = parts[0] * 3600 + parts[1] * 60 + parts[2]
        except (ValueError, IndexError):
            seg_seconds = 0
        if abs(seg_seconds - offset_seconds) < 30:
            return segment.get("speaker_role")
    return "unknown"

def extract_context_snippet(transcript, offset_seconds,
                              window_seconds=10):
    """
    Pull a small window of transcript text around an offset
    to show alongside the structured-field suggestion in the
    review UI. Lets the clinician see the conversational
    context that produced the extraction.
    """
    for segment in transcript.get("segments", []):
        ts = segment.get("timestamp", "00:00:00")
        try:
            parts = [int(p) for p in ts.split(":")]
            seg_seconds = parts[0] * 3600 + parts[1] * 60 + parts[2]
        except (ValueError, IndexError):
            seg_seconds = 0
        if abs(seg_seconds - offset_seconds) < window_seconds:
            return segment.get("text", "")
    return ""

def infer_clinician_action(entity, transcript):
    """
    Heuristic: was the medication mentioned in a context that
    suggests a clinician action (a new prescription, a dose
    adjustment, an order) versus a passing patient mention
    ("I used to take that years ago")? Production uses a
    fine-tuned classifier or an LLM-judge pass; the demo uses
    a simple speaker-role check.
    """
    offset = entity.get("OffsetSeconds", 0)
    speaker_role = lookup_speaker_role_for_offset(
        transcript, offset)
    return speaker_role == "clinician"

def extract_structured_fields(session_context,
                                canonical_transcript):
    """
    Extract clinical entities (medications, conditions) using
    Comprehend Medical with RxNorm and ICD-10 coding, plus
    higher-level fields (orders, follow-up, allergies, vitals,
    referrals) using a Bedrock LLM. Persist all extractions
    for clinician confirmation; never apply them silently to
    the chart.
    """
    session_id = session_context["session_id"]
    encounter_id = session_context["encounter_id"]

    full_text = " ".join(
        seg.get("text", "")
        for seg in canonical_transcript.get("segments", []))

    # Step 5A: extract medications and link to RxNorm.
    rx_response = comprehend_mock.infer_rx_norm(text=full_text)
    coded_medications = []
    for entity in rx_response.get("Entities", []):
        rx_concepts = entity.get("RxNormConcepts", [])
        first_code = (rx_concepts[0]["Code"]
                       if rx_concepts else None)
        offset_seconds = entity.get("OffsetSeconds", 0)
        coded_medications.append({
            "text":           entity.get("Text"),
            "rx_norm_code":   first_code,
            "speaker_role":
                lookup_speaker_role_for_offset(
                    canonical_transcript, offset_seconds),
            "context_snippet":
                extract_context_snippet(
                    canonical_transcript, offset_seconds),
            "clinician_action_likely":
                infer_clinician_action(entity,
                                         canonical_transcript),
            "score":
                Decimal(str(entity.get("Score", 0.85))),
            "clinician_confirmed": False,
        })

    # Step 5A: extract conditions and link to ICD-10.
    icd_response = comprehend_mock.infer_icd10cm(text=full_text)
    coded_conditions = []
    for entity in icd_response.get("Entities", []):
        icd_concepts = entity.get("ICD10CMConcepts", [])
        first_code = (icd_concepts[0]["Code"]
                       if icd_concepts else None)
        offset_seconds = entity.get("OffsetSeconds", 0)
        coded_conditions.append({
            "text":           entity.get("Text"),
            "icd_10_code":    first_code,
            "speaker_role":
                lookup_speaker_role_for_offset(
                    canonical_transcript, offset_seconds),
            "context_snippet":
                extract_context_snippet(
                    canonical_transcript, offset_seconds),
            "score":
                Decimal(str(entity.get("Score", 0.85))),
            "clinician_confirmed": False,
        })

    # Step 5B: use the LLM to identify higher-level structured
    # fields that Comprehend Medical does not directly extract.
    higher_level_response = (
        bedrock_mock.extract_higher_level_fields(
            encounter_id=encounter_id,
            transcript=canonical_transcript))
    higher_level = json.loads(higher_level_response["body"])

    # Mark each higher-level extraction as pending
    # confirmation.
    for category in ("orders_placed", "labs_requested",
                     "imaging_requested",
                     "follow_up_appointments",
                     "patient_reported_vitals",
                     "patient_reported_allergies",
                     "referrals_placed"):
        for item in higher_level.get(category, []):
            item["clinician_confirmed"] = False

    # Step 5C: persist all extractions for clinician
    # confirmation. In production, the PHI-bearing
    # extractions (with context_snippets) are written to
    # a draft-extractions archive in S3 with KMS
    # encryption, and only a reference and counts are
    # stored in the note-state DynamoDB table. The demo
    # stores them directly in the mock state for
    # simplicity.
    structured_extractions = {
        "medications":              coded_medications,
        "conditions":               coded_conditions,
        "orders_placed":
            higher_level.get("orders_placed", []),
        "labs_requested":
            higher_level.get("labs_requested", []),
        "imaging_requested":
            higher_level.get("imaging_requested", []),
        "follow_up_appointments":
            higher_level.get("follow_up_appointments", []),
        "patient_reported_vitals":
            higher_level.get("patient_reported_vitals", []),
        "patient_reported_allergies":
            higher_level.get("patient_reported_allergies", []),
        "referrals_placed":
            higher_level.get("referrals_placed", []),
        "confirmation_status":      "pending_clinician_review",
    }

    note_state.update(session_id, _to_decimal({
        "structured_extractions": structured_extractions,
    }))

    extraction_count = (
        len(coded_medications)
        + len(coded_conditions)
        + len(higher_level.get("orders_placed", []))
        + len(higher_level.get("labs_requested", []))
        + len(higher_level.get("imaging_requested", []))
        + len(higher_level.get("follow_up_appointments", []))
        + len(higher_level.get("patient_reported_vitals", []))
        + len(higher_level.get(
            "patient_reported_allergies", []))
        + len(higher_level.get("referrals_placed", [])))

    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "StructuredExtractionsGenerated",
        extraction_count, "Count",
        dimensions={
            "specialty":
                session_context["clinician_specialty"],
            "language":  session_context["language"],
        })

    audit_log({
        "event_type":         "STRUCTURED_FIELDS_EXTRACTED",
        "session_id":         session_id,
        "medication_count":   len(coded_medications),
        "condition_count":    len(coded_conditions),
        "higher_level_count":
            (extraction_count
             - len(coded_medications)
             - len(coded_conditions)),
        "total_extractions":  extraction_count,
        "timestamp":          _now_iso(),
    })

    return {"extraction_count":       extraction_count,
            "structured_extractions": structured_extractions}

def extract_low_confidence_segments(canonical_transcript,
                                       threshold=ASR_LOW_CONFIDENCE_WORD_THRESHOLD):
    """
    Identify segments with low average word confidence so the
    review UI can highlight them. The clinician's eye is drawn
    to uncertain segments, where ASR errors are most likely.
    """
    flagged = []
    for segment in canonical_transcript.get("segments", []):
        confidence = Decimal(str(
            segment.get("confidence", 1.0)))
        if confidence < threshold:
            flagged.append({
                "timestamp":  segment.get("timestamp"),
                "speaker":    segment.get("speaker_role"),
                "confidence": confidence,
            })
    return flagged

def extract_uncertain_speaker_segments(canonical_transcript):
    """
    Identify segments where speaker attribution is uncertain.
    Production diarization emits per-segment speaker
    confidence; the demo uses a fixture flag.
    """
    uncertain = []
    for segment in canonical_transcript.get("segments", []):
        if segment.get("speaker_uncertain", False):
            uncertain.append({
                "timestamp":     segment.get("timestamp"),
                "current_label": segment.get("speaker_role"),
                "alternatives":
                    segment.get("speaker_alternatives", []),
            })
    return uncertain

def extract_bystander_segments(canonical_transcript,
                                  bystanders):
    """
    Identify segments attributed to a bystander role
    (family_member, caregiver, student, other). The review UI
    surfaces these so the clinician can confirm which
    bystander content belongs in the note.
    """
    if not bystanders:
        return []
    bystander_roles = {"family_member", "caregiver",
                        "student", "other"}
    return [{"timestamp":    seg.get("timestamp"),
             "speaker_role": seg.get("speaker_role"),
             "text_excerpt": (seg.get("text", "") or "")[:80]}
            for seg in canonical_transcript.get("segments", [])
            if seg.get("speaker_role") in bystander_roles]

def clinician_review_request(session_context):
    """
    Assemble the side-by-side review payload for the
    clinician. The web UI uses this to render the draft note
    next to the transcript with click-through citations,
    confidence highlights, and structured-field confirmation
    gates.
    """
    session_id = session_context["session_id"]

    note_record = note_state.get(session_id)
    transcript_object = s3_store.get_object(
        bucket=TRANSCRIPT_BUCKET,
        key=f"{session_id}/canonical_transcript.json")

    if not transcript_object:
        return {"available": False,
                "reason":    "transcript_not_found"}

    canonical_transcript = json.loads(
        transcript_object["body"].decode("utf-8"))

    # If the draft was blocked by faithfulness or guardrails,
    # route the clinician to manual documentation rather than
    # showing the unreliable draft.
    if not note_record.get("draft_available"):
        return {"available":           False,
                "reason":
                    note_record.get("block_reason",
                                     "draft_unavailable"),
                "fallback":             "manual_documentation",
                "canonical_transcript": canonical_transcript}

    review_payload = {
        "available":               True,
        "draft_note":              note_record.get("draft_note"),
        "citations":               note_record.get("citations"),
        "canonical_transcript":    canonical_transcript,
        "structured_extractions":
            note_record.get("structured_extractions"),
        "faithfulness_score":
            note_record.get("faithfulness_score"),
        "faithfulness_severity":
            note_record.get("faithfulness_severity"),
        "faithfulness_annotations":
            note_record.get("faithfulness_annotations"),
        "confidence_highlights":
            extract_low_confidence_segments(
                canonical_transcript),
        "speaker_label_uncertainty":
            extract_uncertain_speaker_segments(
                canonical_transcript),
        "bystander_segments":
            extract_bystander_segments(
                canonical_transcript,
                session_context.get("bystanders", [])),
    }

    audit_log({
        "event_type":      "CLINICIAN_REVIEW_OPENED",
        "session_id":      session_id,
        "faithfulness_severity":
            note_record.get("faithfulness_severity"),
        "extraction_count": (
            len(note_record.get(
                "structured_extractions", {})
                .get("medications", []))
            + len(note_record.get(
                "structured_extractions", {})
                .get("conditions", []))),
        "timestamp":       _now_iso(),
    })

    return review_payload

def clinician_save_review(session_context, review_actions):
    """
    Apply the clinician's edits and confirmation decisions to
    the draft. Each structured-field confirmation is recorded
    with the supporting transcript context. Rejections are
    also recorded to feed the extraction-quality metrics.
    """
    session_id = session_context["session_id"]
    note_record = note_state.get(session_id)

    structured_extractions = note_record.get(
        "structured_extractions", {})

    confirmed_extractions = []
    rejected_extractions = []

    decisions_by_key = {
        (d["category"], d["text"]): d
        for d in review_actions.get("structured_decisions", [])
    }

    for category in ("medications", "conditions",
                     "orders_placed", "labs_requested",
                     "imaging_requested",
                     "follow_up_appointments",
                     "patient_reported_vitals",
                     "patient_reported_allergies",
                     "referrals_placed"):
        for item in structured_extractions.get(category, []):
            key = (category, item.get("text")
                              or item.get("name", ""))
            decision = decisions_by_key.get(key)
            if decision and decision.get("confirmed"):
                item_with_decision = dict(item)
                item_with_decision["clinician_confirmed"] = True
                item_with_decision["category"] = category
                confirmed_extractions.append(item_with_decision)
            else:
                item_with_decision = dict(item)
                item_with_decision["clinician_confirmed"] = False
                item_with_decision["category"] = category
                if decision:
                    item_with_decision["rejection_reason"] = (
                        decision.get("rejection_reason"))
                rejected_extractions.append(item_with_decision)

    edited_note = (review_actions.get("note_edits")
                   or note_record.get("draft_note"))

    note_state.update(session_id, _to_decimal({
        "edited_note":            edited_note,
        "confirmed_extractions":  confirmed_extractions,
        "rejected_extractions":   rejected_extractions,
        "review_completed_at":    _now_iso(),
    }))

    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "ExtractionsConfirmed",
        len(confirmed_extractions), "Count",
        dimensions={
            "specialty":
                session_context["clinician_specialty"],
            "language":  session_context["language"],
        })
    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "ExtractionsRejected",
        len(rejected_extractions), "Count",
        dimensions={
            "specialty":
                session_context["clinician_specialty"],
            "language":  session_context["language"],
        })

    audit_log({
        "event_type":               "CLINICIAN_REVIEW_SAVED",
        "session_id":               session_id,
        "confirmed_extraction_count":
            len(confirmed_extractions),
        "rejected_extraction_count":
            len(rejected_extractions),
        "timestamp":                _now_iso(),
    })

    return {"confirmed": confirmed_extractions,
            "rejected":  rejected_extractions}

def build_idempotency_key(encounter_id, clinician_id,
                            document_type, signed_at):
    """
    Idempotency key for EHR write-back. Truncate signed_at to
    the minute so a within-the-minute retry is treated as the
    same write.
    """
    minute = signed_at.split(":")[0] + ":" + signed_at.split(":")[1]
    raw = f"{encounter_id}|{clinician_id}|{document_type}|{minute}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def clinician_sign(session_context, patient_id):
    """
    Sign the final note, write it to the EHR via FHIR, apply
    the confirmed structured updates to the chart, and
    optionally release the patient-facing summary to the
    portal. Signature is the legal-medical-record boundary;
    after this point any changes are addenda.
    """
    session_id = session_context["session_id"]
    note_record = note_state.get(session_id)

    final_note = (note_record.get("edited_note")
                   or note_record.get("draft_note"))
    confirmed = note_record.get("confirmed_extractions", [])

    signature = clinician_client.get_signature()
    signed_at = signature.get("timestamp", _now_iso())

    idempotency_key = build_idempotency_key(
        encounter_id=session_context["encounter_id"],
        clinician_id=session_context["clinician_id"],
        document_type="clinical_note",
        signed_at=signed_at)

    # Step 6C: write the signed note to the EHR's FHIR
    # DocumentReference resource. Production uses the EHR's
    # FHIR endpoint with SMART on FHIR authentication; the
    # mock just records the write.
    ehr_response = ehr.write_document_reference(
        patient_id=patient_id,
        encounter_id=session_context["encounter_id"],
        document_content=final_note,
        author=session_context["clinician_id"],
        signed_at=signed_at,
        idempotency_key=idempotency_key)

    # Apply confirmed structured-field updates to the chart.
    # Each update is a FHIR resource write (MedicationRequest,
    # Condition, ServiceRequest, etc.).
    for confirmed_item in confirmed:
        category = confirmed_item.get("category")
        if category == "medications":
            ehr.apply_structured_update(
                patient_id=patient_id,
                update_kind="medication",
                payload={
                    "text":         confirmed_item.get("text"),
                    "rx_norm_code":
                        confirmed_item.get("rx_norm_code"),
                    "encounter_id":
                        session_context["encounter_id"],
                })
        elif category == "conditions":
            ehr.apply_structured_update(
                patient_id=patient_id,
                update_kind="problem",
                payload={
                    "text":         confirmed_item.get("text"),
                    "icd_10_code":
                        confirmed_item.get("icd_10_code"),
                    "encounter_id":
                        session_context["encounter_id"],
                })
        elif category in ("labs_requested",
                           "imaging_requested",
                           "orders_placed"):
            ehr.apply_structured_update(
                patient_id=patient_id,
                update_kind="order",
                payload=confirmed_item)
        elif category == "follow_up_appointments":
            ehr.apply_structured_update(
                patient_id=patient_id,
                update_kind="follow_up",
                payload=confirmed_item)
        elif category == "referrals_placed":
            ehr.apply_structured_update(
                patient_id=patient_id,
                update_kind="referral",
                payload=confirmed_item)

    note_state.update(session_id, _to_decimal({
        "signed_note":          final_note,
        "signed_at":            signed_at,
        "signed_by":            session_context["clinician_id"],
        "ehr_document_id":
            ehr_response.get("document_id"),
        "signature":            signature,
        "idempotency_key":      idempotency_key,
        "status":               "signed",
    }))

    started_at_str = session_context["started_at"]
    duration_seconds = (
        datetime.fromisoformat(
            signed_at.replace("Z", "+00:00"))
        - datetime.fromisoformat(
            started_at_str.replace("Z", "+00:00"))
    ).total_seconds()

    event_bus.put_events([{
        "Source":       "ambient_documentation",
        "DetailType":   "note_signed",
        "EventBusName": ENCOUNTER_EVENT_BUS_NAME,
        "Detail": json.dumps({
            "session_id":          session_id,
            "encounter_id":
                session_context["encounter_id"],
            "specialty":
                session_context["clinician_specialty"],
            "ehr_document_id":
                ehr_response.get("document_id"),
            "duration_encounter_to_sign_seconds":
                duration_seconds,
        }),
    }])

    cloudwatch.put_metric(
        CLOUDWATCH_NAMESPACE, "NotesSigned", 1, "Count",
        dimensions={
            "specialty":
                session_context["clinician_specialty"],
            "language":  session_context["language"],
            "visit_type":
                session_context["visit_type"],
        })

    audit_log({
        "event_type":      "NOTE_SIGNED",
        "session_id":      session_id,
        "ehr_document_id":
            ehr_response.get("document_id"),
        "confirmed_extraction_count": len(confirmed),
        "duration_encounter_to_sign_seconds":
            duration_seconds,
        "timestamp":       signed_at,
    })

    return {"signed":          True,
            "ehr_document_id":
                ehr_response.get("document_id"),
            "signed_at":       signed_at,
            "confirmed_extraction_count": len(confirmed)}