$_ CodeCosts

Healthcare and clinical software engineering is the only software discipline where a bug can kill a patient. You are not building a social media feed or an e-commerce checkout — you are writing code that processes protected health information under HIPAA’s Security Rule (§164.312) and Privacy Rule (§164.502), parses HL7v2 pipe-delimited messages where a misread segment can route a lab result to the wrong patient, integrates with EHR systems (Epic, Cerner, MEDITECH) through APIs that each vendor implements differently despite nominally supporting the same FHIR R4 specification, calculates drug dosages where a decimal point error in a weight-based calculation turns a therapeutic dose into a lethal one, and must satisfy FDA 21 CFR Part 11 requirements for electronic records and signatures if your software qualifies as a medical device under the Software as a Medical Device (SaMD) framework. The regulatory surface area is enormous: HIPAA, HITECH, FDA 21 CFR Part 11, IEC 62304 for medical device software lifecycle, ONC Health IT Certification criteria, state-level health data privacy laws (like California’s CMIA), and the emerging EU MDR/IVDR requirements if you sell internationally. A general-purpose AI coding tool trained primarily on web application code does not understand any of this — and “close enough” compliance is not compliance at all.

This guide evaluates every major AI coding tool through the lens of what healthcare and clinical engineers actually build every day. We tested each tool against seven core task areas: HIPAA-compliant data handling (PHI encryption, de-identification, minimum necessary access, audit logging), HL7/FHIR integration (HL7v2 message parsing, FHIR R4 resource construction, terminology binding with SNOMED CT, ICD-10, LOINC, and RxNorm), clinical decision support (drug interaction checking, dosing calculations, CDS Hooks services, clinical alert systems), EHR system integration (SMART on FHIR authorization, Epic/Cerner/generic FHIR adapters, patient context launch), FDA/regulatory compliance (21 CFR Part 11 electronic signatures, IEC 62304 lifecycle, traceability matrices, SOUP documentation), audit trail implementation (immutable logging, tamper detection, retention policies, access anomaly detection), and patient matching and identity (probabilistic matching, Master Patient Index management, duplicate detection, demographic normalization). Every code example in this guide is production-realistic — not a toy hello-world but the actual patterns you will encounter in clinical software systems.

If your work focuses more on genomics and bioinformatics pipelines, see our Bioinformatics Engineers guide. If you are primarily concerned with regulatory automation and policy-as-code across multiple compliance frameworks, see the Compliance Engineers guide. If your focus is security architecture, access control, and threat modeling, see the Security Engineers guide.

Why Healthcare & Clinical Engineering Is Different

Healthcare engineers evaluate AI tools on a fundamentally different axis than application developers. A frontend engineer asks “does this tool write good React?” A healthcare engineer asks “does this tool understand that logging a patient’s name in a stack trace is a HIPAA violation that triggers a breach investigation?” The evaluation criteria are unique to this domain:

• HIPAA compliance is code-level, not just policy. HIPAA is not a checkbox you satisfy once at the infrastructure layer and forget about. The Privacy Rule (§164.502) requires the “minimum necessary” standard — every function that touches protected health information (PHI) must access only the specific data elements required for its purpose. A function that queries a patient record to check insurance eligibility must not also pull diagnosis codes, medication lists, or clinical notes. The Security Rule (§164.312) mandates technical safeguards: access controls (§164.312(a)(1)), audit controls (§164.312(b)), integrity controls (§164.312(c)(1)), person or entity authentication (§164.312(d)), and transmission security (§164.312(e)(1)). Every one of these translates to code. De-identification has two legal methods: Safe Harbor (removing 18 specific identifier types defined in §164.514(b)(2) — names, geographic data smaller than state, dates more specific than year for ages over 89, phone numbers, fax numbers, email addresses, SSNs, MRNs, health plan beneficiary numbers, account numbers, certificate/license numbers, vehicle identifiers, device identifiers and serial numbers, URLs, IP addresses, biometric identifiers, full-face photographs, and any other unique identifying number) and Expert Determination (§164.514(b)(1) requiring a qualified statistical expert to certify re-identification risk is very small). Encryption must be AES-256 at rest and TLS 1.2+ in transit. Breach notification under §164.404 requires notification within 60 days of discovery for breaches affecting 500+ individuals, with immediate notification to HHS and major media outlets in the affected state. Your code must enforce all of this programmatically — not as an afterthought but as a core architectural concern. AI tools that generate database queries pulling SELECT * from patient tables, that log request bodies containing PHI to standard application logs, or that use reversible encoding instead of proper encryption are generating HIPAA violations in every line.
• HL7 and FHIR interoperability is a parsing nightmare. Healthcare data exchange still runs primarily on HL7v2, a protocol designed in 1987 that uses pipe-delimited segments with positional field semantics. An ADT^A01 (admit/visit notification) message contains MSH (message header with sending/receiving facility, message type, encoding characters), EVN (event type with trigger event code and recorded date/time), PID (patient identification with up to 30+ fields including internal ID, external ID, patient name in Last^First^Middle format, mother’s maiden name, date of birth, sex, patient alias, race, address, county code, phone numbers, language, marital status, religion, patient account number, SSN, driver’s license, ethnic group), PV1 (patient visit with admit date, discharge date, attending doctor, referring doctor, consulting doctor, hospital service, temporary location, admit type, VIP indicator, readmission indicator, ambulatory status), and many more segments. Each field uses caret (^) as component separator, ampersand (&) as sub-component separator, tilde (~) for repetition, and backslash (\) as escape character. Z-segments allow custom data that varies by institution. FHIR R4, the modern replacement, defines over 140 resource types — Patient, Observation, MedicationRequest, DiagnosticReport, Condition, Procedure, Encounter, AllergyIntolerance, Immunization, and many more — each with dozens of fields, extensions for custom data, and terminology bindings to code systems like SNOMED CT (over 350,000 concepts), ICD-10-CM (over 72,000 diagnosis codes), LOINC (over 99,000 observation codes), RxNorm (drug terminology linking brand names, generic names, ingredients, and dose forms), and CPT (procedure codes). FHIR search parameters, Bundle transactions (batch and transaction types with different atomicity guarantees), Capability Statements (what a server supports), and the entire SMART on FHIR authorization layer add further complexity. An AI tool that does not understand the difference between HL7v2 field PID-3 (patient identifier list) and PID-5 (patient name), or that confuses FHIR Observation.code (what was measured) with Observation.value (the measurement result), produces integration code that compiles but sends wrong data — which in healthcare means a lab result attached to the wrong patient.
• Clinical data has life-or-death correctness requirements. In 1999, the Mars Climate Orbiter was destroyed because one team used metric units and another used imperial. In healthcare, unit conversion errors happen with drug dosing and they kill patients. Methotrexate dosed daily instead of weekly causes fatal bone marrow suppression. A tenfold error in a pediatric morphine dose — 5 mg instead of 0.5 mg — is lethal. Weight-based dosing requires the system to know whether the weight is in kilograms or pounds, whether it is actual body weight or ideal body weight (relevant for drugs like gentamicin where dosing on actual weight in obese patients causes toxicity), and whether renal function (estimated GFR using CKD-EPI or Cockcroft-Gault equations) requires dose adjustment. Drug-drug interaction checking must classify interactions by severity (contraindicated, serious, moderate, minor) and mechanism (pharmacokinetic via CYP450 enzymes like CYP3A4, CYP2D6, CYP2C19; or pharmacodynamic like additive QT prolongation). Allergy cross-reactivity means a penicillin allergy may or may not contraindicate cephalosporins depending on the generation and the specific side-chain structure. Clinical decision support (CDS) rules implementing CDS Hooks must fire in real time during clinical workflow — a sepsis screening alert that arrives 30 minutes after the clinician has already moved on is useless. Choosing Wisely recommendations must be surfaced as suggestions, not hard stops, because clinical judgment overrides algorithmic recommendations. Every one of these calculations, checks, and alerts is code that must be exactly correct because the consequence of error is patient harm or death, not a 500 error page.
• EHR integration is a legacy minefield. The EHR market is dominated by Epic (used by 38% of US acute care hospitals) and Oracle Health/Cerner (used by 25%), with MEDITECH, Allscripts, athenahealth, and dozens of smaller vendors covering the rest. Each system exposes APIs differently despite the ONC’s 21st Century Cures Act mandate for standardized FHIR R4 access. Epic implements FHIR R4 through its “Open Epic” platform but supplements it with proprietary MyChart APIs, Interconnect web services, and a custom extension ecosystem. Cerner (now Oracle Health) offers its “Millennium” APIs alongside FHIR R4, but its FHIR resources use different extension URIs than Epic’s. Authentication universally uses OAuth 2.0 + SMART on FHIR, but the launch context (EHR launch vs. standalone launch), scopes (patient/*.read, user/*.write, launch/patient), and token refresh behavior differ by vendor. Epic’s App Orchard (now the “Epic on FHIR” Showroom) requires a multi-month review process for production access. Cerner’s Code Console has its own sandbox and approval workflow. CDS Hooks integration — where your application receives notifications during clinical workflow events (patient-view, order-select, order-sign, encounter-start) and returns cards with recommendations — requires real-time response under 500ms or clinicians bypass the system. An AI tool building EHR integration code must understand not just the FHIR specification but the vendor-specific realities: Epic returns extensions with URI http://open.epic.com/FHIR/StructureDefinition/, Cerner uses https://fhir-ehr.cerner.com/r4/StructureDefinition/, and a “generic FHIR client” that ignores these differences will miss critical data fields that exist only in extensions.
• FDA software validation for SaMD (Software as a Medical Device) adds another regulatory layer. If your software provides clinical decision support that is “intended for use in the diagnosis of disease or other conditions, or in the cure, mitigation, treatment, or prevention of disease,” it may be regulated as a medical device under FDA 21 CFR Part 820 (Quality System Regulation). The IEC 62304 standard defines the software development lifecycle: software development planning, software requirements analysis, software architectural design, software detailed design, software unit implementation, software integration and integration testing, software system testing, and software release. Each phase produces documented artifacts. Risk classification determines regulatory burden: Class I (lowest risk, general controls only), Class II (special controls, typically requiring 510(k) clearance demonstrating substantial equivalence to a predicate device), and Class III (highest risk, requiring Premarket Approval (PMA) with clinical trial data). The De Novo pathway covers novel low-to-moderate risk devices without a predicate. 21 CFR Part 11 governs electronic records and electronic signatures: closed system controls (§11.10), open system controls (§11.30), electronic signature requirements (§11.50, §11.70, §11.100), and signature manifestations (§11.200). The Design History File (DHF) must trace every requirement to its implementation, test, and verification. A traceability matrix links user needs to design inputs, design inputs to design outputs, and design outputs to verification and validation results. SOUP (Software of Unknown Provenance) documentation is particularly relevant for AI-generated code — you must document that code’s origin, assess its risk, and verify its behavior. AI tools that generate clinical logic without understanding that every function may need to be traceable back to a software requirement in the DHF produce code that works but cannot pass an FDA audit.
• Audit trails are not optional — they are a federal requirement. HIPAA §164.312(b) mandates audit controls: “Implement hardware, software, and/or procedural mechanisms that record and examine activity in information systems that contain or use electronic protected health information.” This is not a nice-to-have logging feature — it is a legal requirement with civil penalties of $100 to $50,000 per violation (up to $1.5 million per year for each violation category) and criminal penalties up to $250,000 and 10 years imprisonment for knowing misuse. Every access to PHI must be recorded: who accessed it (authenticated user identity), what was accessed (specific patient, specific data elements), when (timestamp with timezone), where (IP address, device, application), why (stated purpose: treatment, payment, healthcare operations, or a specific authorization). The audit log itself must be tamper-evident — if someone modifies the audit log to cover their tracks, that modification must be detectable. Common approaches include hash chaining (each entry includes a hash of the previous entry, creating a blockchain-like chain of integrity), write-once storage (WORM — write once read many), or third-party audit log services with independent integrity verification. Retention requirements are at minimum six years from the date of creation or the last effective date, whichever is later (§164.530(j)). Many organizations retain longer to satisfy state laws or litigation hold requirements. Audit log review is also required — §164.308(a)(1)(ii)(D) requires regular review of records of information system activity such as audit logs, access reports, and security incident tracking reports. AI tools that generate healthcare applications without comprehensive audit logging for every PHI access, without tamper-evident mechanisms, and without retention policy enforcement produce applications that are fundamentally non-compliant from the moment they process their first patient record.
• Patient matching and identity is an unsolved problem. The United States has no universal patient identifier. Congress has included a rider in the HHS appropriations bill every year since 1998 (the Rand Paul rider, originally the McDermott amendment) prohibiting HHS from spending funds to develop a unique patient identifier. This means every healthcare system must match patients across facilities using probabilistic methods based on demographic data — name, date of birth, sex, address, phone number, Social Security number (when available), medical record number (facility-specific) — all of which are imperfect. Names change with marriage. Addresses change with moves. Dates of birth are entered incorrectly (transposed digits, wrong year). The same patient may be “Robert Smith” at one facility and “Bob Smith” at another. The Fellegi-Sunter model (1969) provides the statistical foundation: for each field pair, calculate the probability that the values agree given the records refer to the same person (m-probability) versus the probability they agree by coincidence (u-probability), then combine log-likelihood ratios across fields to produce a composite match score compared against upper (definite match) and lower (definite non-match) thresholds, with scores between the thresholds requiring manual review. Modern approaches use machine learning (random forests, gradient boosting, neural networks) trained on manually reviewed match/non-match pairs, but the training data itself is biased by the demographics of the patient population and the data quality of the participating institutions. Enterprise Master Patient Index (EMPI) systems manage the canonical patient identity across a health system, handling merge operations (when two records are confirmed as the same patient) and unmerge operations (when a merge is discovered to be incorrect). A failed match means a clinician sees an incomplete medical history — missing allergies, missing medications, missing diagnoses. A false match means two different patients’ records are combined — and a clinician acts on information from the wrong patient. Both are patient safety events. AI tools that generate patient lookup code using simple string equality (WHERE last_name = 'Smith' AND first_name = 'John') instead of probabilistic matching algorithms produce systems that fail on the messy, inconsistent, misspelled, abbreviated, culturally diverse names that exist in real patient populations.

Healthcare & Clinical Engineering Task Support Matrix

We tested each tool against seven core healthcare engineering tasks. Ratings reflect real-world performance on clinical-specific prompts, not generic coding ability.

Key observations: Claude Code dominates clinical reasoning tasks because its extended thinking model can hold and cross-reference complex regulatory text, clinical terminology, and multi-step dosing calculations simultaneously. Cursor Pro excels at project-level healthcare code because its codebase indexing lets you query across FHIR resource definitions, EHR adapter implementations, audit logging middleware, and compliance configuration files in a single context. Copilot is useful for routine FHIR resource scaffolding but dangerous for clinical decision logic where it generates plausible-looking code that is clinically incorrect. Tabnine’s strength is code privacy (on-premise deployment), which matters for organizations that cannot send PHI-adjacent code to cloud AI providers, but its healthcare domain knowledge is minimal. Amazon Q has value in AWS-centric health tech stacks (HealthLake, Comprehend Medical) but lacks depth in vendor-specific EHR integration and clinical terminology.

1. HIPAA-Compliant Data Handling

Every healthcare application must enforce HIPAA’s technical safeguards at the code level. This is not about configuring a HIPAA-eligible cloud environment (though that is necessary too) — it is about ensuring that every function, every API endpoint, every background job that touches PHI does so with proper access controls, audit logging, encryption, and minimum necessary filtering. The following PHIAccessGuard class demonstrates the pattern: a wrapper that enforces policy before any PHI access reaches the underlying data store.

import hashlib
import json
import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Optional
from dataclasses import dataclass, field
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
import os

# --- PHI field classification per HIPAA Safe Harbor 18 identifiers ---

class PHICategory(Enum):
    DIRECT_IDENTIFIER = "direct_identifier"      # Names, SSN, MRN
    QUASI_IDENTIFIER = "quasi_identifier"          # DOB, ZIP, dates
    CLINICAL_DATA = "clinical_data"                # Diagnoses, meds, labs
    ADMINISTRATIVE = "administrative"              # Insurance, billing

SAFE_HARBOR_IDENTIFIERS = {
    "patient_name", "street_address", "city", "zip_code",
    "date_of_birth", "admission_date", "discharge_date",
    "phone_number", "fax_number", "email_address",
    "ssn", "mrn", "health_plan_id", "account_number",
    "certificate_number", "vehicle_identifier",
    "device_identifier", "url", "ip_address",
    "biometric_id", "photo", "any_unique_id"
}

@dataclass
class PHIAccessContext:
    """Captures the who/what/when/where/why of every PHI access."""
    user_id: str
    user_role: str
    purpose: str  # treatment, payment, operations, research, patient_request
    patient_id: str
    requested_fields: list[str]
    source_ip: str
    application: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    session_id: str = ""
    break_glass: bool = False  # Emergency override

    def to_audit_record(self) -> dict:
        return {
            "event_type": "phi_access",
            "user_id": self.user_id,
            "user_role": self.user_role,
            "purpose": self.purpose,
            "patient_id": self.patient_id,
            "fields_requested": self.requested_fields,
            "source_ip": self.source_ip,
            "application": self.application,
            "timestamp": self.timestamp.isoformat(),
            "session_id": self.session_id,
            "break_glass": self.break_glass,
        }


# --- Minimum Necessary Access Policy ---

ROLE_FIELD_PERMISSIONS: dict[str, dict[str, set[str]]] = {
    "physician": {
        "treatment": {
            "patient_name", "date_of_birth", "mrn", "sex", "allergies",
            "medications", "diagnoses", "lab_results", "vitals",
            "clinical_notes", "imaging_results", "surgical_history",
        },
        "payment": set(),  # Physicians don't need PHI for billing
    },
    "nurse": {
        "treatment": {
            "patient_name", "date_of_birth", "mrn", "sex", "allergies",
            "medications", "vitals", "clinical_notes", "care_plan",
        },
    },
    "billing_specialist": {
        "payment": {
            "patient_name", "date_of_birth", "mrn", "insurance_id",
            "diagnosis_codes", "procedure_codes", "service_dates",
        },
        "treatment": set(),  # Billing staff get NO clinical detail
    },
    "researcher": {
        "research": {
            # Researchers get de-identified data only
            "age_bucket", "sex", "diagnosis_codes", "lab_results_deidentified",
        },
    },
}


class PHIAccessGuard:
    """
    Enforces HIPAA technical safeguards at the code level.

    - Minimum necessary access (§164.502(b))
    - Audit controls (§164.312(b))
    - Encryption of PHI at rest and in transit
    - De-identification via Safe Harbor method (§164.514(b)(2))
    """

    def __init__(self, audit_logger, encryption_key: bytes):
        self._audit_logger = audit_logger
        self._encryption_key = encryption_key
        self._aesgcm = AESGCM(encryption_key)

    def access_phi(
        self,
        context: PHIAccessContext,
        patient_record: dict[str, Any],
    ) -> dict[str, Any]:
        """
        Gate every PHI access through minimum necessary filtering,
        audit logging, and field-level encryption for sensitive returns.
        """
        # 1. Validate purpose is a recognized TPO category or authorized use
        valid_purposes = {
            "treatment", "payment", "operations",
            "research", "patient_request", "public_health",
        }
        if context.purpose not in valid_purposes:
            self._audit_logger.log_denied_access(context, "invalid_purpose")
            raise PermissionError(
                f"Access denied: '{context.purpose}' is not a valid HIPAA purpose"
            )

        # 2. Apply minimum necessary standard
        permitted_fields = self._get_permitted_fields(
            context.user_role, context.purpose
        )

        # Break-glass override for emergencies — still logged, still audited
        if context.break_glass:
            permitted_fields = set(patient_record.keys())
            self._audit_logger.log_break_glass(context)

        # 3. Filter to minimum necessary
        requested = set(context.requested_fields)
        allowed = requested.intersection(permitted_fields)
        denied = requested - permitted_fields

        if denied and not context.break_glass:
            self._audit_logger.log_partial_denial(context, list(denied))

        # 4. Build filtered response
        filtered_record = {
            field: patient_record[field]
            for field in allowed
            if field in patient_record
        }

        # 5. Log the access (HIPAA §164.312(b))
        self._audit_logger.log_phi_access(
            context=context,
            fields_returned=list(filtered_record.keys()),
            fields_denied=list(denied),
        )

        return filtered_record

    def deidentify_safe_harbor(
        self, record: dict[str, Any]
    ) -> dict[str, Any]:
        """
        Remove all 18 Safe Harbor identifiers per §164.514(b)(2).
        Generalize quasi-identifiers (dates to year, ZIP to 3-digit).
        """
        deidentified = {}

        for field, value in record.items():
            if field in SAFE_HARBOR_IDENTIFIERS:
                continue  # Strip direct identifiers entirely

            if field == "date_of_birth" and value:
                # Generalize to year only; suppress if age > 89
                year = value.year
                age = datetime.now().year - year
                deidentified["age_bucket"] = (
                    "90+" if age > 89 else f"{(age // 10) * 10}-{(age // 10) * 10 + 9}"
                )

            elif field == "zip_code" and value:
                # Truncate to 3-digit prefix; if population < 20k, set to 000
                three_digit = value[:3]
                LOW_POPULATION_ZIPS = {
                    "036", "059", "063", "102", "203",
                    "556", "692", "790", "821", "823",
                    "830", "831", "878", "879", "884",
                    "890", "893",
                }
                deidentified["zip_3digit"] = (
                    "000" if three_digit in LOW_POPULATION_ZIPS else three_digit
                )

            else:
                deidentified[field] = value

        return deidentified

    def encrypt_phi_field(self, plaintext: str) -> bytes:
        """Encrypt a single PHI field using AES-256-GCM (NIST approved)."""
        nonce = os.urandom(12)
        ciphertext = self._aesgcm.encrypt(
            nonce, plaintext.encode("utf-8"), None
        )
        return nonce + ciphertext  # Prepend nonce for decryption

    def decrypt_phi_field(self, encrypted: bytes) -> str:
        """Decrypt a single PHI field."""
        nonce = encrypted[:12]
        ciphertext = encrypted[12:]
        plaintext = self._aesgcm.decrypt(nonce, ciphertext, None)
        return plaintext.decode("utf-8")

    def _get_permitted_fields(self, role: str, purpose: str) -> set[str]:
        role_perms = ROLE_FIELD_PERMISSIONS.get(role, {})
        return role_perms.get(purpose, set())

What AI tools get right: Claude Code generates the role-based minimum necessary filtering correctly and cites the specific HIPAA sections. Cursor Pro catches PHI field leaks when you have the access guard in one file and the API endpoint in another because it indexes both. What they get wrong: Copilot and Windsurf routinely generate code that logs the full patient record to application logs for debugging — a direct HIPAA violation. Copilot also tends to use base64 encoding instead of actual encryption, which satisfies no regulatory requirement. All tools struggle with the Safe Harbor de-identification edge cases: the low-population ZIP code suppression (you must zero out 3-digit ZIP prefixes where the population is less than 20,000) and the age >89 generalization requirement.

2. HL7/FHIR Integration

Healthcare interoperability remains the most technically demanding aspect of clinical software. You must parse HL7v2 messages that have been in production since the 1990s and simultaneously build FHIR R4 resources for modern API integrations. The following example demonstrates both: an HL7v2 ADT^A01 parser that extracts patient demographics, and a FHIR R4 resource builder that constructs compliant Patient and Observation resources with proper terminology binding.

import re
from datetime import datetime
from typing import Any, Optional
from dataclasses import dataclass, field
import json


# --- HL7v2 Message Parser ---

@dataclass
class HL7v2Field:
    """Represents a parsed HL7v2 field with components and sub-components."""
    components: list[list[str]]  # components[i] = sub-components

    @property
    def value(self) -> str:
        """Return the first component's first sub-component (most common access)."""
        if self.components and self.components[0]:
            return self.components[0][0]
        return ""

    def component(self, index: int) -> str:
        """Get a specific component (1-indexed per HL7v2 convention)."""
        idx = index - 1
        if idx < len(self.components) and self.components[idx]:
            return self.components[idx][0]
        return ""

    def subcomponent(self, comp_idx: int, sub_idx: int) -> str:
        """Get a specific sub-component (both 1-indexed)."""
        ci, si = comp_idx - 1, sub_idx - 1
        if ci < len(self.components) and si < len(self.components[ci]):
            return self.components[ci][si]
        return ""


class HL7v2Parser:
    """
    Parses HL7v2 pipe-delimited messages with proper escape handling.

    Handles:
    - Field separator (|), component (^), sub-component (&),
      repetition (~), escape (\\)
    - HL7v2 escape sequences: \\F\\ (|), \\S\\ (^), \\T\\ (&),
      \\R\\ (~), \\E\\ (\\), \\X0D\\ (hex), \\.br\\ (line break)
    - Z-segments (custom/institution-specific segments)
    """

    ESCAPE_MAP = {
        "F": "|",
        "S": "^",
        "T": "&",
        "R": "~",
        "E": "\\",
    }

    def __init__(self, raw_message: str):
        # Normalize line endings
        self._raw = raw_message.replace("\r\n", "\r").replace("\n", "\r")
        self._segments: dict[str, list[list[HL7v2Field]]] = {}
        self._parse()

    def _unescape(self, value: str) -> str:
        """Process HL7v2 escape sequences."""
        result = value
        for code, replacement in self.ESCAPE_MAP.items():
            result = result.replace(f"\\{code}\\", replacement)
        # Handle hex escapes like \X0D\
        result = re.sub(
            r"\\X([0-9A-Fa-f]+)\\",
            lambda m: bytes.fromhex(m.group(1)).decode("ascii", errors="replace"),
            result,
        )
        # Handle .br line breaks
        result = result.replace("\\.br\\", "\n")
        return result

    def _parse(self):
        segments_raw = self._raw.strip().split("\r")
        for segment_str in segments_raw:
            if len(segment_str) < 3:
                continue

            segment_name = segment_str[:3]

            # MSH is special: MSH-1 IS the field separator
            if segment_name == "MSH":
                fields_raw = segment_str[4:].split("|")
                fields_raw.insert(0, "|")  # MSH-1 = field separator
            else:
                fields_raw = segment_str[4:].split("|") if len(segment_str) > 3 else []

            fields = []
            for field_raw in fields_raw:
                # Handle repetitions (separated by ~)
                repetitions = field_raw.split("~")
                # Parse first repetition's components
                components = []
                for comp_str in repetitions[0].split("^"):
                    sub_components = [
                        self._unescape(sc) for sc in comp_str.split("&")
                    ]
                    components.append(sub_components)
                fields.append(HL7v2Field(components=components))

            if segment_name not in self._segments:
                self._segments[segment_name] = []
            self._segments[segment_name].append(fields)

    def get_segment(self, name: str, index: int = 0) -> Optional[list[HL7v2Field]]:
        """Get a segment by name (supports repeating segments via index)."""
        segments = self._segments.get(name, [])
        return segments[index] if index < len(segments) else None

    def get_field(self, segment: str, field_num: int, seg_index: int = 0) -> Optional[HL7v2Field]:
        """Get a specific field (1-indexed per HL7v2 convention)."""
        seg = self.get_segment(segment, seg_index)
        if seg is None:
            return None
        idx = field_num - 1
        return seg[idx] if idx < len(seg) else None

    def get_patient_name(self) -> dict[str, str]:
        """Extract patient name from PID-5 (XPN data type)."""
        name_field = self.get_field("PID", 5)
        if name_field is None:
            return {}
        return {
            "family": name_field.component(1),
            "given": name_field.component(2),
            "middle": name_field.component(3),
            "suffix": name_field.component(4),
            "prefix": name_field.component(5),
            "degree": name_field.component(6),
        }

    def get_patient_identifiers(self) -> list[dict[str, str]]:
        """Extract patient identifiers from PID-3 (CX data type)."""
        id_field = self.get_field("PID", 3)
        if id_field is None:
            return []
        return [{
            "id": id_field.component(1),
            "assigning_authority": id_field.component(4),
            "identifier_type": id_field.component(5),
        }]

    def get_message_type(self) -> tuple[str, str]:
        """Return (message_type, trigger_event) from MSH-9."""
        msg_type = self.get_field("MSH", 9)
        if msg_type is None:
            return ("", "")
        return (msg_type.component(1), msg_type.component(2))


# --- FHIR R4 Resource Builder ---

class FHIRResourceBuilder:
    """
    Builds FHIR R4-compliant resources with proper terminology binding,
    extensions, and validation.
    """

    # Standard code systems
    SNOMED_CT = "http://snomed.info/sct"
    ICD10_CM = "http://hl7.org/fhir/sid/icd-10-cm"
    LOINC = "http://loinc.org"
    RXNORM = "http://www.nlm.nih.gov/research/umls/rxnorm"
    CPT = "http://www.ama-assn.org/go/cpt"
    FHIR_OBSERVATION_CATEGORY = "http://terminology.hl7.org/CodeSystem/observation-category"
    UCUM = "http://unitsofmeasure.org"

    @staticmethod
    def build_patient(
        mrn: str,
        family_name: str,
        given_names: list[str],
        birth_date: str,
        gender: str,
        identifiers: Optional[list[dict]] = None,
        address: Optional[dict] = None,
        telecom: Optional[list[dict]] = None,
        race_code: Optional[str] = None,
        ethnicity_code: Optional[str] = None,
    ) -> dict[str, Any]:
        """
        Build a FHIR R4 Patient resource with US Core profile extensions
        for race and ethnicity.
        """
        resource: dict[str, Any] = {
            "resourceType": "Patient",
            "meta": {
                "profile": [
                    "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"
                ]
            },
            "identifier": [
                {
                    "use": "usual",
                    "type": {
                        "coding": [{
                            "system": "http://terminology.hl7.org/CodeSystem/v2-0203",
                            "code": "MR",
                            "display": "Medical Record Number",
                        }]
                    },
                    "system": "urn:oid:1.2.3.4.5.6",  # Replace with facility OID
                    "value": mrn,
                }
            ],
            "name": [{
                "use": "official",
                "family": family_name,
                "given": given_names,
            }],
            "gender": gender,  # male | female | other | unknown
            "birthDate": birth_date,  # YYYY-MM-DD
        }

        if identifiers:
            for ident in identifiers:
                resource["identifier"].append(ident)

        if address:
            resource["address"] = [{
                "use": "home",
                "type": "physical",
                "line": address.get("lines", []),
                "city": address.get("city", ""),
                "state": address.get("state", ""),
                "postalCode": address.get("postal_code", ""),
                "country": address.get("country", "US"),
            }]

        if telecom:
            resource["telecom"] = telecom

        # US Core requires race and ethnicity extensions
        extensions = []
        if race_code:
            extensions.append({
                "url": "http://hl7.org/fhir/us/core/StructureDefinition/us-core-race",
                "extension": [{
                    "url": "ombCategory",
                    "valueCoding": {
                        "system": "urn:oid:2.16.840.1.113883.6.238",
                        "code": race_code,
                    }
                }, {
                    "url": "text",
                    "valueString": race_code,
                }]
            })

        if ethnicity_code:
            extensions.append({
                "url": "http://hl7.org/fhir/us/core/StructureDefinition/us-core-ethnicity",
                "extension": [{
                    "url": "ombCategory",
                    "valueCoding": {
                        "system": "urn:oid:2.16.840.1.113883.6.238",
                        "code": ethnicity_code,
                    }
                }, {
                    "url": "text",
                    "valueString": ethnicity_code,
                }]
            })

        if extensions:
            resource["extension"] = extensions

        return resource

    @staticmethod
    def build_observation(
        patient_reference: str,
        loinc_code: str,
        loinc_display: str,
        value: float,
        unit: str,
        ucum_code: str,
        effective_datetime: str,
        status: str = "final",
        category_code: str = "laboratory",
        encounter_reference: Optional[str] = None,
        reference_range_low: Optional[float] = None,
        reference_range_high: Optional[float] = None,
    ) -> dict[str, Any]:
        """
        Build a FHIR R4 Observation resource with LOINC coding,
        UCUM units, and optional reference ranges.
        """
        observation: dict[str, Any] = {
            "resourceType": "Observation",
            "meta": {
                "profile": [
                    "http://hl7.org/fhir/us/core/StructureDefinition/us-core-observation-lab"
                ]
            },
            "status": status,
            "category": [{
                "coding": [{
                    "system": FHIRResourceBuilder.FHIR_OBSERVATION_CATEGORY,
                    "code": category_code,
                    "display": category_code.capitalize(),
                }]
            }],
            "code": {
                "coding": [{
                    "system": FHIRResourceBuilder.LOINC,
                    "code": loinc_code,
                    "display": loinc_display,
                }]
            },
            "subject": {"reference": patient_reference},
            "effectiveDateTime": effective_datetime,
            "valueQuantity": {
                "value": value,
                "unit": unit,
                "system": FHIRResourceBuilder.UCUM,
                "code": ucum_code,
            },
        }

        if encounter_reference:
            observation["encounter"] = {"reference": encounter_reference}

        if reference_range_low is not None or reference_range_high is not None:
            ref_range: dict[str, Any] = {}
            if reference_range_low is not None:
                ref_range["low"] = {
                    "value": reference_range_low,
                    "unit": unit,
                    "system": FHIRResourceBuilder.UCUM,
                    "code": ucum_code,
                }
            if reference_range_high is not None:
                ref_range["high"] = {
                    "value": reference_range_high,
                    "unit": unit,
                    "system": FHIRResourceBuilder.UCUM,
                    "code": ucum_code,
                }
            observation["referenceRange"] = [ref_range]

        return observation

    @staticmethod
    def build_bundle_transaction(entries: list[dict[str, Any]]) -> dict[str, Any]:
        """
        Build a FHIR Bundle of type 'transaction' for atomic operations.
        Each entry must succeed or the entire bundle is rolled back.
        """
        bundle_entries = []
        for entry in entries:
            resource_type = entry.get("resourceType", "")
            bundle_entries.append({
                "fullUrl": f"urn:uuid:{entry.get('id', '')}",
                "resource": entry,
                "request": {
                    "method": "POST",
                    "url": resource_type,
                }
            })

        return {
            "resourceType": "Bundle",
            "type": "transaction",
            "entry": bundle_entries,
        }

What AI tools get right: Cursor Pro handles FHIR resource building well when you already have FHIR resource files in your project — it learns your extension patterns and autocompletes consistently. Claude Code generates correct HL7v2 parsing including escape sequences, which most tools miss entirely. What they get wrong: Every tool except Claude Code initially misses the HL7v2 escape sequence handling — the \F\, \S\, \T\, \R\, \E\ sequences that replace special characters within field values. Copilot and Windsurf consistently forget the MSH segment’s special parsing rule (MSH-1 is the field separator character itself, not parsed like other fields). For FHIR, the most common error across all tools is omitting the meta.profile element that declares US Core conformance — without it, FHIR validators reject the resource. All tools also struggle with FHIR terminology binding strength: some code systems are required bindings (you must use a code from the specified ValueSet), while others are extensible (you should use a code from the ValueSet but can use others if none fit), and tools do not distinguish between the two.

3. Clinical Decision Support

Clinical decision support (CDS) is where software directly influences patient care. A CDS system checks drug interactions, validates dosing, screens for sepsis, surfaces clinical guidelines, and alerts clinicians to potential safety issues. The stakes are absolute: a false negative (missing a dangerous interaction) can cause patient harm, while excessive false positives (“alert fatigue”) cause clinicians to ignore all alerts including critical ones. The following ClinicalDecisionEngine demonstrates drug interaction checking with severity classification, weight-based dosing with renal adjustment, and a CDS Hooks-compatible service interface.

from dataclasses import dataclass
from enum import Enum
from typing import Any, Optional
from decimal import Decimal, ROUND_HALF_UP
import math


class InteractionSeverity(Enum):
    CONTRAINDICATED = "contraindicated"  # Must not co-prescribe
    SERIOUS = "serious"                  # Use only if benefit outweighs risk
    MODERATE = "moderate"                # Monitor patient closely
    MINOR = "minor"                      # Minimal clinical significance


class InteractionMechanism(Enum):
    PHARMACOKINETIC_CYP3A4 = "CYP3A4 inhibition/induction"
    PHARMACOKINETIC_CYP2D6 = "CYP2D6 inhibition/induction"
    PHARMACOKINETIC_CYP2C19 = "CYP2C19 inhibition/induction"
    PHARMACOKINETIC_PGLYCOPROTEIN = "P-glycoprotein interaction"
    PHARMACODYNAMIC_QT = "Additive QT prolongation"
    PHARMACODYNAMIC_SEROTONIN = "Serotonin syndrome risk"
    PHARMACODYNAMIC_BLEEDING = "Additive bleeding risk"
    PHARMACODYNAMIC_CNS = "Additive CNS depression"
    PHARMACODYNAMIC_NEPHROTOXIC = "Additive nephrotoxicity"
    PHARMACODYNAMIC_HEPATOTOXIC = "Additive hepatotoxicity"


@dataclass
class DrugInteraction:
    drug_a_rxnorm: str
    drug_a_name: str
    drug_b_rxnorm: str
    drug_b_name: str
    severity: InteractionSeverity
    mechanism: InteractionMechanism
    clinical_consequence: str
    management: str
    evidence_level: str  # "established", "probable", "suspected", "possible"


@dataclass
class DosingRecommendation:
    drug_name: str
    calculated_dose: Decimal
    dose_unit: str
    frequency: str
    route: str
    max_single_dose: Optional[Decimal] = None
    max_daily_dose: Optional[Decimal] = None
    renal_adjustment: Optional[str] = None
    hepatic_adjustment: Optional[str] = None
    warnings: list[str] = None

    def __post_init__(self):
        if self.warnings is None:
            self.warnings = []


@dataclass
class CDSCard:
    """CDS Hooks response card — shown to clinician in EHR workflow."""
    summary: str
    detail: str
    indicator: str  # "critical", "warning", "info"
    source_label: str
    source_url: Optional[str] = None
    suggestions: list[dict] = None
    selection_behavior: str = "at-most-one"

    def to_cds_hooks_card(self) -> dict:
        card = {
            "summary": self.summary,
            "detail": self.detail,
            "indicator": self.indicator,
            "source": {"label": self.source_label},
        }
        if self.source_url:
            card["source"]["url"] = self.source_url
        if self.suggestions:
            card["suggestions"] = self.suggestions
            card["selectionBehavior"] = self.selection_behavior
        return card


class ClinicalDecisionEngine:
    """
    Production clinical decision support engine.

    Implements:
    - Drug-drug interaction checking with severity classification
    - Weight-based dosing with renal/hepatic adjustment
    - CDS Hooks service interface for EHR integration
    - Clinical alert management with fatigue reduction
    """

    def __init__(self, interaction_db: list[DrugInteraction]):
        # Index interactions by RxNorm code pairs for O(1) lookup
        self._interactions: dict[tuple[str, str], DrugInteraction] = {}
        for interaction in interaction_db:
            key_ab = (interaction.drug_a_rxnorm, interaction.drug_b_rxnorm)
            key_ba = (interaction.drug_b_rxnorm, interaction.drug_a_rxnorm)
            self._interactions[key_ab] = interaction
            self._interactions[key_ba] = interaction

    def check_interactions(
        self,
        new_drug_rxnorm: str,
        current_medications: list[dict[str, str]],
    ) -> list[DrugInteraction]:
        """
        Check a new drug against all current medications.
        Returns interactions sorted by severity (most severe first).
        """
        found_interactions = []
        for med in current_medications:
            key = (new_drug_rxnorm, med["rxnorm"])
            interaction = self._interactions.get(key)
            if interaction is not None:
                found_interactions.append(interaction)

        # Sort: contraindicated first, then serious, moderate, minor
        severity_order = {
            InteractionSeverity.CONTRAINDICATED: 0,
            InteractionSeverity.SERIOUS: 1,
            InteractionSeverity.MODERATE: 2,
            InteractionSeverity.MINOR: 3,
        }
        found_interactions.sort(key=lambda x: severity_order[x.severity])
        return found_interactions

    def calculate_weight_based_dose(
        self,
        drug_name: str,
        dose_per_kg: Decimal,
        dose_unit: str,
        patient_weight_kg: Decimal,
        frequency: str,
        route: str,
        max_single_dose: Optional[Decimal] = None,
        max_daily_dose: Optional[Decimal] = None,
        egfr: Optional[float] = None,
        renal_dose_adjustments: Optional[dict] = None,
        use_ideal_body_weight: bool = False,
        patient_height_cm: Optional[Decimal] = None,
        patient_sex: Optional[str] = None,
    ) -> DosingRecommendation:
        """
        Calculate weight-based dose with safety checks.

        Uses actual body weight by default. For drugs with narrow
        therapeutic index in obese patients (e.g., aminoglycosides),
        set use_ideal_body_weight=True.
        """
        warnings = []

        # Calculate ideal body weight if needed (Devine formula)
        effective_weight = patient_weight_kg
        if use_ideal_body_weight and patient_height_cm and patient_sex:
            height_inches = float(patient_height_cm) / Decimal("2.54")
            if patient_sex == "male":
                ibw = Decimal("50") + Decimal("2.3") * Decimal(str(max(0, height_inches - 60)))
            else:
                ibw = Decimal("45.5") + Decimal("2.3") * Decimal(str(max(0, height_inches - 60)))
            effective_weight = ibw
            warnings.append(
                f"Using ideal body weight ({ibw:.1f} kg) instead of "
                f"actual weight ({patient_weight_kg:.1f} kg)"
            )

        # Calculate base dose
        calculated_dose = (dose_per_kg * effective_weight).quantize(
            Decimal("0.01"), rounding=ROUND_HALF_UP
        )

        # Apply max single dose cap
        if max_single_dose and calculated_dose > max_single_dose:
            warnings.append(
                f"Calculated dose {calculated_dose} {dose_unit} exceeds "
                f"max single dose {max_single_dose} {dose_unit}. Capped."
            )
            calculated_dose = max_single_dose

        # Renal dose adjustment based on eGFR
        renal_note = None
        if egfr is not None and renal_dose_adjustments:
            for threshold_str, adjustment in sorted(
                renal_dose_adjustments.items(), key=lambda x: float(x[0])
            ):
                threshold = float(threshold_str)
                if egfr < threshold:
                    if adjustment["type"] == "reduce_percent":
                        reduction = Decimal(str(adjustment["value"])) / Decimal("100")
                        original_dose = calculated_dose
                        calculated_dose = (calculated_dose * (1 - reduction)).quantize(
                            Decimal("0.01"), rounding=ROUND_HALF_UP
                        )
                        renal_note = (
                            f"Dose reduced {adjustment['value']}% for eGFR "
                            f"{egfr:.0f} mL/min/1.73m2 "
                            f"(from {original_dose} to {calculated_dose} {dose_unit})"
                        )
                    elif adjustment["type"] == "extend_interval":
                        frequency = adjustment["new_frequency"]
                        renal_note = (
                            f"Interval extended to {frequency} for eGFR "
                            f"{egfr:.0f} mL/min/1.73m2"
                        )
                    break

            if renal_note:
                warnings.append(renal_note)

        return DosingRecommendation(
            drug_name=drug_name,
            calculated_dose=calculated_dose,
            dose_unit=dose_unit,
            frequency=frequency,
            route=route,
            max_single_dose=max_single_dose,
            max_daily_dose=max_daily_dose,
            renal_adjustment=renal_note,
            warnings=warnings,
        )

    def estimate_gfr_ckd_epi(
        self,
        serum_creatinine: float,
        age: int,
        sex: str,
        race: Optional[str] = None,
    ) -> float:
        """
        Estimate GFR using the 2021 CKD-EPI equation (race-free).

        The 2021 revision removed the race coefficient, following
        the NKF-ASN Task Force recommendation.

        Returns eGFR in mL/min/1.73m2.
        """
        # 2021 CKD-EPI equation (race-free)
        if sex == "female":
            kappa = 0.7
            alpha = -0.241 if serum_creatinine <= 0.7 else -1.2
            sex_coefficient = 1.012
        else:
            kappa = 0.9
            alpha = -0.302 if serum_creatinine <= 0.9 else -1.2
            sex_coefficient = 1.0

        egfr = (
            142
            * (min(serum_creatinine / kappa, 1.0) ** alpha)
            * (max(serum_creatinine / kappa, 1.0) ** -1.2)
            * (0.9938 ** age)
            * sex_coefficient
        )

        return round(egfr, 1)

    def generate_order_sign_cards(
        self,
        hook_context: dict[str, Any],
    ) -> list[dict]:
        """
        CDS Hooks 'order-sign' service.

        Receives the hook context containing the proposed order and
        the patient's current medications, and returns CDS cards for
        any clinical concerns.
        """
        cards = []
        proposed_med = hook_context.get("context", {}).get("draftOrders", {})
        current_meds = hook_context.get("prefetch", {}).get("currentMedications", [])

        if not proposed_med or not current_meds:
            return []

        # Extract RxNorm code from proposed medication
        proposed_rxnorm = ""
        proposed_name = ""
        for entry in proposed_med.get("entry", []):
            resource = entry.get("resource", {})
            if resource.get("resourceType") == "MedicationRequest":
                codings = (
                    resource.get("medicationCodeableConcept", {})
                    .get("coding", [])
                )
                for coding in codings:
                    if coding.get("system") == "http://www.nlm.nih.gov/research/umls/rxnorm":
                        proposed_rxnorm = coding["code"]
                        proposed_name = coding.get("display", "")
                        break

        if not proposed_rxnorm:
            return []

        # Build current medication list from prefetch
        current_med_list = []
        for entry in current_meds.get("entry", []):
            resource = entry.get("resource", {})
            if resource.get("resourceType") == "MedicationRequest":
                codings = (
                    resource.get("medicationCodeableConcept", {})
                    .get("coding", [])
                )
                for coding in codings:
                    if coding.get("system") == "http://www.nlm.nih.gov/research/umls/rxnorm":
                        current_med_list.append({
                            "rxnorm": coding["code"],
                            "name": coding.get("display", ""),
                        })
                        break

        # Check interactions
        interactions = self.check_interactions(proposed_rxnorm, current_med_list)

        for interaction in interactions:
            if interaction.severity == InteractionSeverity.CONTRAINDICATED:
                indicator = "critical"
            elif interaction.severity == InteractionSeverity.SERIOUS:
                indicator = "warning"
            else:
                indicator = "info"

            card = CDSCard(
                summary=(
                    f"{interaction.severity.value.upper()}: "
                    f"{interaction.drug_a_name} + {interaction.drug_b_name}"
                ),
                detail=(
                    f"Mechanism: {interaction.mechanism.value}. "
                    f"{interaction.clinical_consequence} "
                    f"Management: {interaction.management} "
                    f"(Evidence: {interaction.evidence_level})"
                ),
                indicator=indicator,
                source_label="Clinical Decision Support",
                suggestions=[{
                    "label": f"Cancel {proposed_name} order",
                    "actions": [{
                        "type": "delete",
                        "description": f"Remove {proposed_name} from draft orders",
                    }]
                }] if interaction.severity == InteractionSeverity.CONTRAINDICATED else None,
            )
            cards.append(card.to_cds_hooks_card())

        return cards

What AI tools get right: Claude Code generates the 2021 CKD-EPI equation correctly (race-free version, per the NKF-ASN Task Force recommendation) and understands the clinical significance of using the updated formula. It also correctly structures CDS Hooks response cards with the indicator field mapped to clinical severity. Cursor produces good drug interaction data structures when it can see existing interaction models in your project. What they get wrong: Copilot generates drug interaction checking without severity classification — it returns “interaction found: true/false” without distinguishing between a contraindicated combination (warfarin + aspirin in high doses) and a minor interaction that requires only monitoring. All tools except Claude Code use the older CKD-EPI equation with the race coefficient, which was deprecated in 2021. Windsurf and Copilot generate dosing calculations using floating-point arithmetic instead of Decimal, which produces rounding errors that are unacceptable in medication dosing (0.1 + 0.2 != 0.3 in IEEE 754). No tool reliably generates the ideal body weight calculation for drugs that require it, and none spontaneously add the max single dose cap that prevents a 150 kg patient from receiving a dangerous dose calculated purely from weight.

4. EHR System Integration

Integrating with Electronic Health Record systems is the bread and butter of clinical software development, and it is significantly more complex than integrating with a standard REST API. Every EHR vendor implements FHIR R4 slightly differently, authentication requires the SMART on FHIR protocol (an OAuth 2.0 profile specific to healthcare), and production access requires multi-month vendor review processes. The following EHRIntegrationClient demonstrates the SMART on FHIR authorization flow, vendor-specific adapter patterns, and the retry/rate-limiting logic that production EHR integrations require.

import time
import secrets
import hashlib
import base64
import urllib.parse
from typing import Any, Optional, Protocol
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
import json
import logging

logger = logging.getLogger(__name__)


@dataclass
class SMARTAuthConfig:
    """SMART on FHIR OAuth 2.0 configuration."""
    client_id: str
    client_secret: Optional[str]  # Confidential clients only
    redirect_uri: str
    scopes: list[str]  # e.g., ["patient/*.read", "launch/patient", "openid", "fhirUser"]
    fhir_base_url: str
    authorize_url: str = ""
    token_url: str = ""

    @classmethod
    def from_well_known(cls, fhir_base_url: str, **kwargs) -> "SMARTAuthConfig":
        """
        Discover auth endpoints from .well-known/smart-configuration
        per SMART App Launch Framework.
        """
        # In production, fetch from {fhir_base_url}/.well-known/smart-configuration
        # This returns authorize_url, token_url, scopes_supported, etc.
        config = cls(fhir_base_url=fhir_base_url, **kwargs)
        return config


@dataclass
class TokenResponse:
    access_token: str
    token_type: str
    expires_in: int
    refresh_token: Optional[str]
    scope: str
    patient: Optional[str]  # Patient ID from launch context
    id_token: Optional[str]  # OpenID Connect
    issued_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    @property
    def is_expired(self) -> bool:
        elapsed = (datetime.now(timezone.utc) - self.issued_at).total_seconds()
        return elapsed >= (self.expires_in - 60)  # 60s buffer


class FHIRClient(Protocol):
    """Protocol for vendor-agnostic FHIR operations."""
    def read(self, resource_type: str, resource_id: str) -> dict: ...
    def search(self, resource_type: str, params: dict) -> dict: ...
    def create(self, resource: dict) -> dict: ...
    def update(self, resource: dict) -> dict: ...


class EHRIntegrationClient:
    """
    Multi-vendor EHR integration client with SMART on FHIR auth,
    vendor-specific adapters, retry logic, and rate limiting.

    Supports:
    - Epic (Open Epic / FHIR R4)
    - Cerner / Oracle Health (Millennium FHIR R4)
    - Generic FHIR R4 servers
    """

    # Vendor-specific rate limits (requests per second)
    RATE_LIMITS = {
        "epic": 10,        # Epic throttles aggressively
        "cerner": 20,      # Cerner is more generous
        "generic": 50,     # Default for unknown servers
    }

    # Vendor-specific extension base URIs
    EXTENSION_BASES = {
        "epic": "http://open.epic.com/FHIR/StructureDefinition/",
        "cerner": "https://fhir-ehr.cerner.com/r4/StructureDefinition/",
    }

    def __init__(
        self,
        auth_config: SMARTAuthConfig,
        vendor: str = "generic",
        max_retries: int = 3,
        retry_backoff_base: float = 1.0,
    ):
        self._auth_config = auth_config
        self._vendor = vendor
        self._max_retries = max_retries
        self._retry_backoff_base = retry_backoff_base
        self._token: Optional[TokenResponse] = None
        self._last_request_time: float = 0.0
        self._rate_limit = self.RATE_LIMITS.get(vendor, self.RATE_LIMITS["generic"])
        self._min_request_interval = 1.0 / self._rate_limit

    def build_authorization_url(self, launch_token: Optional[str] = None) -> tuple[str, str]:
        """
        Build SMART on FHIR authorization URL.

        Returns (authorization_url, state) for redirect.
        Supports both EHR launch (with launch_token) and standalone launch.
        """
        state = secrets.token_urlsafe(32)

        # PKCE (Proof Key for Code Exchange) — required by SMART App Launch 2.0
        code_verifier = secrets.token_urlsafe(64)
        code_challenge = base64.urlsafe_b64encode(
            hashlib.sha256(code_verifier.encode("ascii")).digest()
        ).rstrip(b"=").decode("ascii")

        params = {
            "response_type": "code",
            "client_id": self._auth_config.client_id,
            "redirect_uri": self._auth_config.redirect_uri,
            "scope": " ".join(self._auth_config.scopes),
            "state": state,
            "aud": self._auth_config.fhir_base_url,
            "code_challenge": code_challenge,
            "code_challenge_method": "S256",
        }

        if launch_token:
            params["launch"] = launch_token  # EHR launch context

        auth_url = (
            f"{self._auth_config.authorize_url}?"
            f"{urllib.parse.urlencode(params)}"
        )

        # Store code_verifier for token exchange (in production, use session store)
        self._code_verifier = code_verifier

        return auth_url, state

    def exchange_code_for_token(self, authorization_code: str) -> TokenResponse:
        """
        Exchange authorization code for access token.
        Includes PKCE code_verifier for SMART App Launch 2.0.
        """
        token_params = {
            "grant_type": "authorization_code",
            "code": authorization_code,
            "redirect_uri": self._auth_config.redirect_uri,
            "client_id": self._auth_config.client_id,
            "code_verifier": self._code_verifier,
        }

        # In production, POST to self._auth_config.token_url
        # response = httpx.post(self._auth_config.token_url, data=token_params)
        # token_data = response.json()

        # The response includes patient context from EHR launch
        # token_data["patient"] = "Patient/12345"
        # This is how your app knows which patient the clinician selected

        # Placeholder for actual HTTP call
        token_data = {}  # Replace with actual HTTP response

        self._token = TokenResponse(
            access_token=token_data.get("access_token", ""),
            token_type=token_data.get("token_type", "Bearer"),
            expires_in=token_data.get("expires_in", 3600),
            refresh_token=token_data.get("refresh_token"),
            scope=token_data.get("scope", ""),
            patient=token_data.get("patient"),
            id_token=token_data.get("id_token"),
        )
        return self._token

    def _ensure_valid_token(self):
        """Refresh token if expired."""
        if self._token is None:
            raise RuntimeError("Not authenticated. Call exchange_code_for_token first.")
        if self._token.is_expired and self._token.refresh_token:
            self._refresh_token()

    def _refresh_token(self):
        """Refresh the access token using refresh_token grant."""
        refresh_params = {
            "grant_type": "refresh_token",
            "refresh_token": self._token.refresh_token,
            "client_id": self._auth_config.client_id,
        }
        # In production: POST to token_url with refresh_params
        # Update self._token with new response
        pass

    def _rate_limit_wait(self):
        """Enforce vendor-specific rate limiting."""
        now = time.monotonic()
        elapsed = now - self._last_request_time
        if elapsed < self._min_request_interval:
            time.sleep(self._min_request_interval - elapsed)
        self._last_request_time = time.monotonic()

    def _request_with_retry(
        self,
        method: str,
        url: str,
        headers: dict,
        json_body: Optional[dict] = None,
    ) -> dict:
        """
        Execute HTTP request with retry logic for transient failures.

        Handles:
        - 429 Too Many Requests (respect Retry-After header)
        - 500/502/503/504 (transient server errors)
        - Connection timeouts
        - ETag/If-Match for optimistic concurrency
        """
        self._rate_limit_wait()

        for attempt in range(self._max_retries + 1):
            try:
                # In production, use httpx or requests
                # response = httpx.request(method, url, headers=headers, json=json_body)

                # Placeholder response handling:
                status_code = 200  # Replace with actual response
                response_json = {}  # Replace with actual response

                if status_code == 200 or status_code == 201:
                    return response_json

                if status_code == 429:
                    # Respect Retry-After header
                    retry_after = 5  # Parse from response headers
                    logger.warning(
                        f"Rate limited by {self._vendor} EHR. "
                        f"Waiting {retry_after}s (attempt {attempt + 1})"
                    )
                    time.sleep(retry_after)
                    continue

                if status_code == 409:
                    # Conflict — resource version mismatch
                    raise ConflictError(
                        f"Resource version conflict. Re-read and retry with "
                        f"current ETag. (FHIR uses If-Match for optimistic locking)"
                    )

                if status_code in (500, 502, 503, 504):
                    backoff = self._retry_backoff_base * (2 ** attempt)
                    logger.warning(
                        f"Server error {status_code} from {self._vendor} EHR. "
                        f"Retrying in {backoff}s (attempt {attempt + 1})"
                    )
                    time.sleep(backoff)
                    continue

                raise EHRAPIError(f"Unexpected status {status_code} from {self._vendor} EHR")

            except ConnectionError:
                if attempt < self._max_retries:
                    backoff = self._retry_backoff_base * (2 ** attempt)
                    logger.warning(f"Connection error. Retrying in {backoff}s")
                    time.sleep(backoff)
                else:
                    raise

        raise EHRAPIError(f"Max retries ({self._max_retries}) exceeded for {self._vendor} EHR")

    def read_patient(self, patient_id: str) -> dict:
        """Read a Patient resource, handling vendor-specific extensions."""
        self._ensure_valid_token()

        url = f"{self._auth_config.fhir_base_url}/Patient/{patient_id}"
        headers = {
            "Authorization": f"Bearer {self._token.access_token}",
            "Accept": "application/fhir+json",
        }

        patient = self._request_with_retry("GET", url, headers)

        # Normalize vendor-specific extensions into a standard format
        if self._vendor == "epic":
            patient = self._normalize_epic_extensions(patient)
        elif self._vendor == "cerner":
            patient = self._normalize_cerner_extensions(patient)

        return patient

    def search_observations(
        self,
        patient_id: str,
        category: Optional[str] = None,
        code: Optional[str] = None,
        date_from: Optional[str] = None,
        date_to: Optional[str] = None,
    ) -> list[dict]:
        """
        Search Observations with pagination handling.
        FHIR servers return paginated Bundle responses via 'next' links.
        """
        self._ensure_valid_token()

        params = {"patient": patient_id, "_count": "100"}
        if category:
            params["category"] = category
        if code:
            params["code"] = code
        if date_from:
            params["date"] = f"ge{date_from}"
        if date_to:
            params["date"] = params.get("date", "") + f"&date=le{date_to}" if "date" in params else f"le{date_to}"

        all_entries = []
        url = f"{self._auth_config.fhir_base_url}/Observation?{urllib.parse.urlencode(params)}"
        headers = {
            "Authorization": f"Bearer {self._token.access_token}",
            "Accept": "application/fhir+json",
        }

        while url:
            bundle = self._request_with_retry("GET", url, headers)
            entries = bundle.get("entry", [])
            all_entries.extend([e.get("resource", {}) for e in entries])

            # Follow pagination links
            url = None
            for link in bundle.get("link", []):
                if link.get("relation") == "next":
                    url = link["url"]
                    break

        return all_entries

    def _normalize_epic_extensions(self, resource: dict) -> dict:
        """
        Extract Epic-specific extensions into a normalized format.
        Epic uses extensions for data not in base FHIR (e.g., MyChart status).
        """
        epic_base = self.EXTENSION_BASES["epic"]
        normalized_extensions = {}

        for ext in resource.get("extension", []):
            url = ext.get("url", "")
            if url.startswith(epic_base):
                ext_name = url[len(epic_base):]
                normalized_extensions[f"epic_{ext_name}"] = ext

        resource["_vendor_extensions"] = normalized_extensions
        return resource

    def _normalize_cerner_extensions(self, resource: dict) -> dict:
        """Extract Cerner-specific extensions into a normalized format."""
        cerner_base = self.EXTENSION_BASES["cerner"]
        normalized_extensions = {}

        for ext in resource.get("extension", []):
            url = ext.get("url", "")
            if url.startswith(cerner_base):
                ext_name = url[len(cerner_base):]
                normalized_extensions[f"cerner_{ext_name}"] = ext

        resource["_vendor_extensions"] = normalized_extensions
        return resource


class ConflictError(Exception):
    pass


class EHRAPIError(Exception):
    pass

What AI tools get right: Cursor Pro excels here because EHR integration involves many files (OAuth config, FHIR client, vendor adapters, token management) and its multi-file indexing helps maintain consistency. Claude Code correctly generates the SMART on FHIR authorization flow including PKCE (Proof Key for Code Exchange), which became required in SMART App Launch 2.0 — most other tools generate OAuth 2.0 without PKCE. What they get wrong: Every tool except Claude Code generates a simple OAuth 2.0 flow without the SMART-specific aud parameter (the FHIR server URL that the token is scoped to) and without the launch parameter for EHR launch context. Copilot frequently generates FHIR search code without pagination handling — FHIR servers return paginated Bundles, and a search for all of a patient’s observations may span multiple pages. No tool spontaneously adds vendor-specific rate limiting — Epic in particular aggressively throttles applications that exceed their rate limits, and getting rate-limited during a clinical workflow means a clinician is staring at a spinner while a patient waits. Windsurf and Tabnine generate generic REST clients that treat FHIR as a regular JSON API, missing the application/fhir+json content type and the FHIR-specific error handling (OperationOutcome resources).

5. FDA/Regulatory Compliance

If your software qualifies as a Software as a Medical Device (SaMD), you are subject to FDA oversight and must maintain a quality management system with documented evidence of every development activity. 21 CFR Part 11 governs electronic records and signatures. IEC 62304 defines the software development lifecycle. The following FDAComplianceManager demonstrates electronic signature implementation, traceability matrix generation, SOUP documentation, and Design History File entry management — the artifacts that FDA auditors actually ask for.

import hashlib
import json
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from dataclasses import dataclass, field
from enum import Enum


class RiskClassification(Enum):
    CLASS_I = "Class I"      # General controls
    CLASS_II = "Class II"    # Special controls + 510(k)
    CLASS_III = "Class III"  # Premarket approval (PMA)


class SoftwareUnitCategory(Enum):
    """IEC 62304 software safety classification."""
    CLASS_A = "A"  # No injury or damage to health
    CLASS_B = "B"  # Non-serious injury
    CLASS_C = "C"  # Death or serious injury possible


class VerificationStatus(Enum):
    NOT_STARTED = "not_started"
    IN_PROGRESS = "in_progress"
    PASSED = "passed"
    FAILED = "failed"
    DEFERRED = "deferred"


@dataclass
class ElectronicSignature:
    """
    21 CFR Part 11 compliant electronic signature.

    Requirements (§11.50, §11.70, §11.100, §11.200):
    - Signed by the individual executing the signature
    - Include the printed name, date/time, and meaning (review, approval, authorship)
    - Be linked to the signed record such that the signature cannot be
      transplanted to another record
    - Use at least two identification components (e.g., user ID + password)
    """
    signer_id: str
    signer_name: str
    meaning: str  # "review", "approval", "authorship", "verification"
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    signature_id: str = field(default_factory=lambda: str(uuid.uuid4()))

    def bind_to_record(self, record_content: str) -> dict:
        """
        Create a signature binding that links this signature to a specific
        record content hash, preventing signature transplantation.
        """
        content_hash = hashlib.sha256(record_content.encode("utf-8")).hexdigest()
        binding = {
            "signature_id": self.signature_id,
            "signer_id": self.signer_id,
            "signer_name": self.signer_name,
            "meaning": self.meaning,
            "timestamp": self.timestamp.isoformat(),
            "record_hash": content_hash,
            "binding_hash": hashlib.sha256(
                f"{self.signature_id}:{content_hash}:{self.signer_id}".encode("utf-8")
            ).hexdigest(),
        }
        return binding


@dataclass
class SoftwareRequirement:
    """A traceable software requirement for the DHF."""
    req_id: str              # e.g., "SRS-CDS-001"
    title: str
    description: str
    source: str              # User need or design input that originated this
    risk_classification: SoftwareUnitCategory
    verification_method: str  # "test", "inspection", "analysis", "demonstration"
    verification_status: VerificationStatus = VerificationStatus.NOT_STARTED
    test_case_ids: list[str] = field(default_factory=list)
    implementation_refs: list[str] = field(default_factory=list)  # file:function refs
    parent_req_id: Optional[str] = None  # For requirement decomposition


@dataclass
class SOUPComponent:
    """
    Software of Unknown Provenance documentation.

    IEC 62304 §8.1.2 requires documentation of all SOUP items including:
    - Title, manufacturer, version
    - Identified anomalies relevant to safety
    - Functional and performance requirements
    """
    title: str
    manufacturer: str
    version: str
    purpose: str
    functional_requirements: list[str]
    performance_requirements: list[str]
    known_anomalies: list[str]
    risk_assessment: str
    verification_method: str
    is_ai_generated: bool = False  # Flag for AI-generated code requiring extra scrutiny


class FDAComplianceManager:
    """
    Manages FDA regulatory compliance artifacts for SaMD development.

    Covers:
    - 21 CFR Part 11 electronic records and signatures
    - IEC 62304 software lifecycle documentation
    - Design History File (DHF) management
    - Traceability matrix generation
    - SOUP component documentation
    """

    def __init__(self, product_name: str, risk_class: RiskClassification):
        self._product_name = product_name
        self._risk_class = risk_class
        self._requirements: dict[str, SoftwareRequirement] = {}
        self._soup_components: list[SOUPComponent] = []
        self._signatures: list[dict] = []
        self._dhf_entries: list[dict] = []

    def add_requirement(self, req: SoftwareRequirement):
        """Add a software requirement to the tracked set."""
        self._requirements[req.req_id] = req

    def add_soup_component(self, soup: SOUPComponent):
        """Register a SOUP component with its documentation."""
        self._soup_components.append(soup)

    def sign_record(
        self,
        record_content: str,
        signer_id: str,
        signer_name: str,
        meaning: str,
        auth_verified: bool = False,
    ) -> dict:
        """
        Apply a 21 CFR Part 11 compliant electronic signature.

        auth_verified must be True, indicating the caller has already
        verified the signer's identity with two identification components
        (§11.200: e.g., user ID + password, biometric + PIN).
        """
        if not auth_verified:
            raise PermissionError(
                "21 CFR Part 11 §11.200 requires signature execution to be "
                "preceded by two-factor identification. Set auth_verified=True "
                "only after verifying signer identity."
            )

        signature = ElectronicSignature(
            signer_id=signer_id,
            signer_name=signer_name,
            meaning=meaning,
        )

        binding = signature.bind_to_record(record_content)
        self._signatures.append(binding)
        return binding

    def generate_traceability_matrix(self) -> dict[str, Any]:
        """
        Generate a requirements traceability matrix for the DHF.

        Maps: User Need -> Design Input -> Software Requirement ->
              Implementation -> Test Case -> Verification Result

        This is the primary artifact FDA auditors examine to verify
        that every requirement is implemented and verified.
        """
        matrix_rows = []
        untraced_requirements = []
        untested_requirements = []

        for req_id, req in self._requirements.items():
            row = {
                "requirement_id": req.req_id,
                "title": req.title,
                "source": req.source,
                "risk_class": req.risk_classification.value,
                "implementation_refs": req.implementation_refs,
                "test_case_ids": req.test_case_ids,
                "verification_method": req.verification_method,
                "verification_status": req.verification_status.value,
                "is_fully_traced": bool(req.implementation_refs and req.test_case_ids),
            }
            matrix_rows.append(row)

            if not req.implementation_refs:
                untraced_requirements.append(req_id)
            if not req.test_case_ids:
                untested_requirements.append(req_id)

        # Calculate coverage metrics
        total = len(self._requirements)
        implemented = total - len(untraced_requirements)
        tested = total - len(untested_requirements)
        verified_pass = sum(
            1 for r in self._requirements.values()
            if r.verification_status == VerificationStatus.PASSED
        )

        matrix = {
            "product": self._product_name,
            "risk_classification": self._risk_class.value,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "summary": {
                "total_requirements": total,
                "implemented": implemented,
                "tested": tested,
                "verified_passed": verified_pass,
                "implementation_coverage": f"{implemented / total * 100:.1f}%" if total else "N/A",
                "test_coverage": f"{tested / total * 100:.1f}%" if total else "N/A",
                "verification_rate": f"{verified_pass / total * 100:.1f}%" if total else "N/A",
            },
            "gaps": {
                "unimplemented_requirements": untraced_requirements,
                "untested_requirements": untested_requirements,
            },
            "rows": matrix_rows,
        }

        return matrix

    def generate_soup_report(self) -> dict[str, Any]:
        """
        Generate SOUP documentation report per IEC 62304 §8.1.2.

        Special attention to AI-generated code components, which
        require additional risk assessment and verification.
        """
        ai_generated = [s for s in self._soup_components if s.is_ai_generated]
        third_party = [s for s in self._soup_components if not s.is_ai_generated]

        report = {
            "product": self._product_name,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "summary": {
                "total_soup_components": len(self._soup_components),
                "ai_generated_components": len(ai_generated),
                "third_party_components": len(third_party),
            },
            "ai_generated_code_notice": (
                "Components flagged as AI-generated require additional "
                "verification per IEC 62304 §5.5.3 (software unit verification). "
                "AI-generated code is treated as SOUP because its internal logic "
                "cannot be fully inspected or guaranteed by the manufacturer. "
                "Each AI-generated component must have: (1) documented prompt or "
                "specification that generated it, (2) independent code review, "
                "(3) unit tests covering all identified risk scenarios, "
                "(4) static analysis results."
            ),
            "components": [],
        }

        for soup in self._soup_components:
            entry = {
                "title": soup.title,
                "manufacturer": soup.manufacturer,
                "version": soup.version,
                "purpose": soup.purpose,
                "is_ai_generated": soup.is_ai_generated,
                "functional_requirements": soup.functional_requirements,
                "performance_requirements": soup.performance_requirements,
                "known_anomalies": soup.known_anomalies,
                "risk_assessment": soup.risk_assessment,
                "verification_method": soup.verification_method,
            }
            report["components"].append(entry)

        return report

    def create_dhf_entry(
        self,
        entry_type: str,
        title: str,
        content: str,
        author_id: str,
        author_name: str,
        related_requirements: list[str],
        auth_verified: bool = False,
    ) -> dict:
        """
        Create a Design History File entry with electronic signature.

        Entry types: "design_input", "design_output", "verification",
                     "validation", "review", "change_order", "risk_analysis"
        """
        entry = {
            "entry_id": str(uuid.uuid4()),
            "entry_type": entry_type,
            "title": title,
            "content": content,
            "related_requirements": related_requirements,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "product": self._product_name,
        }

        # Sign the entry
        entry_json = json.dumps(entry, sort_keys=True)
        signature = self.sign_record(
            record_content=entry_json,
            signer_id=author_id,
            signer_name=author_name,
            meaning="authorship",
            auth_verified=auth_verified,
        )
        entry["signature"] = signature

        self._dhf_entries.append(entry)
        return entry

    def assess_ai_generated_code_risk(
        self,
        component_name: str,
        ai_tool_used: str,
        prompt_or_spec: str,
        generated_code_hash: str,
        safety_classification: SoftwareUnitCategory,
    ) -> dict:
        """
        Risk assessment for AI-generated code in SaMD context.

        FDA's 2023 guidance on AI/ML in SaMD requires documentation of:
        - The AI tool and version used
        - The input (prompt/specification) that generated the code
        - The output (generated code, identified by hash)
        - Risk assessment specific to the safety classification
        - Additional verification measures applied
        """
        risk_mitigations = {
            SoftwareUnitCategory.CLASS_A: [
                "Standard code review by qualified developer",
                "Unit tests with branch coverage >= 80%",
                "Static analysis (no critical findings)",
            ],
            SoftwareUnitCategory.CLASS_B: [
                "Independent code review by two qualified developers",
                "Unit tests with branch coverage >= 90%",
                "Static analysis (no critical or high findings)",
                "Integration testing against clinical scenarios",
                "Boundary value testing for all clinical calculations",
            ],
            SoftwareUnitCategory.CLASS_C: [
                "Independent code review by two qualified developers + clinical SME",
                "Unit tests with branch coverage >= 95%",
                "Static analysis (no findings above informational)",
                "Full integration testing against clinical scenarios",
                "Boundary and stress testing for all clinical calculations",
                "Clinical validation with representative patient data",
                "Formal verification of safety-critical logic where feasible",
                "Manual line-by-line code inspection documented in DHF",
            ],
        }

        assessment = {
            "component_name": component_name,
            "ai_tool_used": ai_tool_used,
            "prompt_hash": hashlib.sha256(prompt_or_spec.encode("utf-8")).hexdigest(),
            "generated_code_hash": generated_code_hash,
            "safety_classification": safety_classification.value,
            "assessed_at": datetime.now(timezone.utc).isoformat(),
            "required_mitigations": risk_mitigations.get(
                safety_classification,
                risk_mitigations[SoftwareUnitCategory.CLASS_C],
            ),
            "soup_classification": (
                "AI-generated code is classified as SOUP per IEC 62304. "
                "The manufacturer cannot fully guarantee the internal logic "
                "of AI-generated code and must apply verification measures "
                "commensurate with the software safety classification."
            ),
            "regulatory_notes": [
                "21 CFR Part 820.30(g) — Design validation shall ensure "
                "that devices conform to defined user needs and intended uses.",
                "IEC 62304 §5.5.3 — Software unit verification shall demonstrate "
                "that the software unit meets its detailed design.",
                "IEC 62304 §8.1.2 — SOUP identification and documentation.",
            ],
        }

        return assessment

What AI tools get right: Claude Code understands the 21 CFR Part 11 electronic signature requirements (two-factor identification, signature meaning, record binding) and generates the traceability matrix structure that FDA auditors actually request. It also raises the important issue of AI-generated code as SOUP — a critical regulatory concern for any team using AI coding tools to build medical device software. What they get wrong: No tool except Claude Code spontaneously addresses the SOUP classification for AI-generated code, which is arguably the most important regulatory question for teams using AI tools in SaMD development. Copilot generates “digital signature” implementations using cryptographic signatures (RSA/ECDSA) instead of the regulatory concept of an electronic signature per Part 11, which is about identity verification and meaning, not cryptographic proof. Windsurf produces generic document versioning instead of the specific Design History File structure that IEC 62304 requires. All tools struggle with traceability — the concept that every requirement must trace forward to implementation and tests, and backward to user needs — because traceability is a project-level concern that spans many files and cannot be captured in a single code snippet.

6. Audit Trail Implementation

Healthcare audit trails are not application logging. They are federally mandated records of every interaction with protected health information, required to be tamper-evident, retained for at minimum six years, and available for review by HHS Office for Civil Rights investigators, internal compliance officers, and legal discovery. The following HealthcareAuditLogger implements hash-chained immutable logging, PHI access event recording, access anomaly detection, and retention policy enforcement.

import hashlib
import json
import uuid
from datetime import datetime, timezone, timedelta
from typing import Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict


class AuditEventType(Enum):
    PHI_ACCESS = "phi_access"
    PHI_CREATE = "phi_create"
    PHI_UPDATE = "phi_update"
    PHI_DELETE = "phi_delete"
    PHI_EXPORT = "phi_export"
    PHI_DISCLOSURE = "phi_disclosure"  # Sharing PHI outside the entity
    LOGIN_SUCCESS = "login_success"
    LOGIN_FAILURE = "login_failure"
    PERMISSION_CHANGE = "permission_change"
    BREAK_GLASS = "break_glass"        # Emergency access override
    SYSTEM_ACCESS = "system_access"     # Automated/system-level PHI access
    AUDIT_LOG_ACCESS = "audit_log_access"  # Meta: accessing the audit log itself


class AnomalyType(Enum):
    AFTER_HOURS_ACCESS = "after_hours_access"
    EXCESSIVE_RECORDS = "excessive_records"
    UNUSUAL_DEPARTMENT = "unusual_department"
    VIP_PATIENT_ACCESS = "vip_patient_access"
    BREAK_GLASS_FREQUENCY = "break_glass_frequency"
    FAILED_LOGIN_BURST = "failed_login_burst"
    MASS_EXPORT = "mass_export"


@dataclass
class AuditEntry:
    """
    A single audit log entry per HIPAA §164.312(b).
    Captures the who/what/when/where/why of every PHI interaction.
    """
    event_id: str
    event_type: AuditEventType
    timestamp: datetime
    user_id: str
    user_role: str
    user_department: str
    patient_id: Optional[str]
    resource_type: str          # "Patient", "Observation", "MedicationRequest", etc.
    resource_id: Optional[str]
    action: str                  # "read", "write", "delete", "search", "export"
    fields_accessed: list[str]   # Specific PHI fields accessed
    purpose: str                 # "treatment", "payment", "operations", etc.
    source_ip: str
    source_application: str
    outcome: str                 # "success", "denied", "error"
    detail: str                  # Human-readable description
    previous_hash: str           # Hash of the previous entry (chain integrity)
    entry_hash: str = ""         # Computed hash of this entry

    def compute_hash(self) -> str:
        """Compute SHA-256 hash of this entry for chain integrity."""
        hashable_content = json.dumps({
            "event_id": self.event_id,
            "event_type": self.event_type.value,
            "timestamp": self.timestamp.isoformat(),
            "user_id": self.user_id,
            "patient_id": self.patient_id,
            "resource_type": self.resource_type,
            "resource_id": self.resource_id,
            "action": self.action,
            "fields_accessed": self.fields_accessed,
            "purpose": self.purpose,
            "outcome": self.outcome,
            "previous_hash": self.previous_hash,
        }, sort_keys=True)
        return hashlib.sha256(hashable_content.encode("utf-8")).hexdigest()


class HealthcareAuditLogger:
    """
    HIPAA-compliant audit logger with tamper detection.

    Implements:
    - Hash-chained immutable log entries (blockchain-like integrity)
    - PHI access event recording with full context
    - Access anomaly detection for proactive breach identification
    - Retention policy enforcement (§164.530(j): 6 year minimum)
    - Audit log access logging (meta-auditing)

    The hash chain ensures that if any entry is modified or deleted,
    the chain breaks and the tampering is detectable.
    """

    GENESIS_HASH = "0" * 64  # Initial hash for the first entry in the chain

    def __init__(
        self,
        storage_backend: Any,  # Database, object store, WORM storage
        anomaly_callback: Optional[Callable] = None,
        retention_years: int = 6,
    ):
        self._storage = storage_backend
        self._anomaly_callback = anomaly_callback
        self._retention_years = retention_years
        self._last_hash = self.GENESIS_HASH
        self._access_counts: dict[str, dict[str, int]] = defaultdict(
            lambda: defaultdict(int)
        )

    def log_phi_access(
        self,
        context: dict[str, Any],
        fields_returned: list[str],
        fields_denied: list[str],
    ) -> AuditEntry:
        """Log a PHI access event with full context."""
        entry = self._create_entry(
            event_type=AuditEventType.PHI_ACCESS,
            user_id=context["user_id"],
            user_role=context["user_role"],
            user_department=context.get("department", "unknown"),
            patient_id=context.get("patient_id"),
            resource_type=context.get("resource_type", "Patient"),
            resource_id=context.get("resource_id"),
            action="read",
            fields_accessed=fields_returned,
            purpose=context.get("purpose", "treatment"),
            source_ip=context.get("source_ip", "unknown"),
            source_application=context.get("application", "unknown"),
            outcome="success" if fields_returned else "denied",
            detail=(
                f"Accessed {len(fields_returned)} fields. "
                f"Denied {len(fields_denied)} fields: {fields_denied}"
                if fields_denied else
                f"Accessed {len(fields_returned)} fields"
            ),
        )

        # Track for anomaly detection
        date_key = entry.timestamp.strftime("%Y-%m-%d")
        self._access_counts[entry.user_id][date_key] += 1

        # Check for anomalies
        self._check_anomalies(entry)

        return entry

    def log_break_glass(self, context: dict[str, Any]) -> AuditEntry:
        """
        Log an emergency break-glass access override.

        Break-glass access bypasses normal access controls for
        emergencies but triggers immediate review.
        """
        entry = self._create_entry(
            event_type=AuditEventType.BREAK_GLASS,
            user_id=context["user_id"],
            user_role=context["user_role"],
            user_department=context.get("department", "unknown"),
            patient_id=context.get("patient_id"),
            resource_type="Patient",
            resource_id=context.get("patient_id"),
            action="break_glass_read",
            fields_accessed=["ALL_FIELDS"],
            purpose="emergency",
            source_ip=context.get("source_ip", "unknown"),
            source_application=context.get("application", "unknown"),
            outcome="success",
            detail=(
                f"BREAK GLASS: Emergency access override by "
                f"{context['user_id']} ({context['user_role']}). "
                f"Reason: {context.get('reason', 'not specified')}. "
                f"All PHI fields accessed for patient {context.get('patient_id')}."
            ),
        )

        # Break-glass always triggers immediate notification
        if self._anomaly_callback:
            self._anomaly_callback(
                AnomalyType.BREAK_GLASS_FREQUENCY,
                entry,
                "Break-glass emergency access invoked — requires supervisory review within 24 hours",
            )

        return entry

    def log_denied_access(self, context: dict[str, Any], reason: str) -> AuditEntry:
        """Log a denied PHI access attempt."""
        return self._create_entry(
            event_type=AuditEventType.PHI_ACCESS,
            user_id=context.get("user_id", "unknown"),
            user_role=context.get("user_role", "unknown"),
            user_department=context.get("department", "unknown"),
            patient_id=context.get("patient_id"),
            resource_type=context.get("resource_type", "Patient"),
            resource_id=context.get("resource_id"),
            action="read",
            fields_accessed=[],
            purpose=context.get("purpose", "unknown"),
            source_ip=context.get("source_ip", "unknown"),
            source_application=context.get("application", "unknown"),
            outcome="denied",
            detail=f"Access denied: {reason}",
        )

    def log_partial_denial(self, context: dict[str, Any], denied_fields: list[str]) -> AuditEntry:
        """Log when some requested fields were denied (minimum necessary enforcement)."""
        return self._create_entry(
            event_type=AuditEventType.PHI_ACCESS,
            user_id=context.get("user_id", "unknown"),
            user_role=context.get("user_role", "unknown"),
            user_department=context.get("department", "unknown"),
            patient_id=context.get("patient_id"),
            resource_type="Patient",
            resource_id=context.get("patient_id"),
            action="read",
            fields_accessed=[],
            purpose=context.get("purpose", "treatment"),
            source_ip=context.get("source_ip", "unknown"),
            source_application=context.get("application", "unknown"),
            outcome="partial_denial",
            detail=f"Minimum necessary enforcement: denied fields {denied_fields}",
        )

    def verify_chain_integrity(self, entries: list[AuditEntry]) -> dict[str, Any]:
        """
        Verify the hash chain integrity of a sequence of audit entries.

        If any entry has been modified, deleted, or inserted out of order,
        the chain will break at that point.
        """
        if not entries:
            return {"valid": True, "entries_checked": 0, "breaks": []}

        breaks = []
        expected_previous = self.GENESIS_HASH

        for i, entry in enumerate(entries):
            # Verify this entry's hash matches its content
            computed = entry.compute_hash()
            if computed != entry.entry_hash:
                breaks.append({
                    "index": i,
                    "event_id": entry.event_id,
                    "type": "content_tampered",
                    "detail": (
                        f"Entry hash mismatch: stored={entry.entry_hash}, "
                        f"computed={computed}"
                    ),
                })

            # Verify chain linkage
            if entry.previous_hash != expected_previous:
                breaks.append({
                    "index": i,
                    "event_id": entry.event_id,
                    "type": "chain_break",
                    "detail": (
                        f"Chain break: expected previous_hash={expected_previous}, "
                        f"found={entry.previous_hash}"
                    ),
                })

            expected_previous = entry.entry_hash

        return {
            "valid": len(breaks) == 0,
            "entries_checked": len(entries),
            "breaks": breaks,
            "verified_at": datetime.now(timezone.utc).isoformat(),
        }

    def enforce_retention_policy(self, current_date: Optional[datetime] = None) -> dict[str, Any]:
        """
        Enforce the minimum retention period per HIPAA §164.530(j).

        Records must be retained for AT LEAST 6 years from creation or
        last effective date. This method identifies records eligible for
        deletion and records that must be retained.

        IMPORTANT: Even after the retention period, records under
        litigation hold must not be deleted.
        """
        now = current_date or datetime.now(timezone.utc)
        retention_cutoff = now - timedelta(days=self._retention_years * 365)

        # In production, query the storage backend
        # eligible_for_deletion = storage.query(timestamp < retention_cutoff)
        # under_litigation_hold = storage.query(litigation_hold=True)

        return {
            "retention_years": self._retention_years,
            "cutoff_date": retention_cutoff.isoformat(),
            "policy": (
                f"Records created before {retention_cutoff.date()} are "
                f"eligible for deletion per the {self._retention_years}-year "
                f"retention policy, UNLESS subject to litigation hold, "
                f"ongoing investigation, or state law requiring longer retention."
            ),
            "warning": (
                "Before deleting any audit records, verify: "
                "(1) No active litigation hold, "
                "(2) No ongoing OCR investigation, "
                "(3) State retention laws are satisfied, "
                "(4) Deletion is itself audited and approved by Privacy Officer."
            ),
        }

    def _check_anomalies(self, entry: AuditEntry):
        """
        Detect access anomalies that may indicate unauthorized access.

        Common HIPAA breach indicators:
        - After-hours access to patient records
        - Accessing records outside normal department scope
        - Unusually high volume of record access
        - VIP patient record snooping
        - Repeated break-glass usage
        """
        if not self._anomaly_callback:
            return

        # After-hours access (11 PM - 5 AM local time)
        hour = entry.timestamp.hour
        if hour >= 23 or hour < 5:
            if entry.user_role not in ("physician", "nurse", "emergency_physician"):
                self._anomaly_callback(
                    AnomalyType.AFTER_HOURS_ACCESS,
                    entry,
                    f"After-hours PHI access by {entry.user_role} at {entry.timestamp.strftime('%H:%M')}",
                )

        # Excessive record access (> 50 patients in a day for non-batch roles)
        date_key = entry.timestamp.strftime("%Y-%m-%d")
        daily_count = self._access_counts[entry.user_id][date_key]
        if daily_count > 50 and entry.user_role not in ("system", "batch_processor", "report_generator"):
            self._anomaly_callback(
                AnomalyType.EXCESSIVE_RECORDS,
                entry,
                f"User {entry.user_id} accessed {daily_count} patient records today",
            )

    def _create_entry(self, **kwargs) -> AuditEntry:
        """Create a hash-chained audit entry."""
        entry = AuditEntry(
            event_id=str(uuid.uuid4()),
            timestamp=datetime.now(timezone.utc),
            previous_hash=self._last_hash,
            **kwargs,
        )
        entry.entry_hash = entry.compute_hash()
        self._last_hash = entry.entry_hash

        # Persist the entry (in production, write to WORM storage or append-only DB)
        # self._storage.append(entry)

        return entry

What AI tools get right: Claude Code generates the hash-chained audit log pattern correctly and understands the HIPAA-specific requirements (six-year retention, tamper-evidence, access anomaly detection). Cursor Pro helps maintain audit logging consistency across a codebase by detecting when a new endpoint accesses PHI but does not call the audit logger. What they get wrong: Copilot generates standard application logging (Python logging module) without tamper-evident mechanisms — a standard log file can be edited by anyone with filesystem access, which defeats the purpose of audit controls. Windsurf adds structured logging (JSON format) but misses the hash chain integrity requirement. All tools except Claude Code omit the meta-auditing requirement: access to the audit log itself must be logged, creating a recursive audit requirement that is easy to miss. No tool spontaneously adds the break-glass pattern (emergency access override with mandatory post-hoc review), which is a standard feature in clinical systems and a common audit focus area.

7. Patient Matching & Identity

Patient matching is one of the hardest problems in healthcare IT, and it is critically important for patient safety. Without a universal patient identifier, healthcare systems must match patients across facilities using imperfect demographic data. The following PatientMatcher implements the Fellegi-Sunter probabilistic matching model, demographic normalization, a duplicate detection scorer, and MPI merge/unmerge operations.

import math
import re
from typing import Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import date


class MatchDecision(Enum):
    MATCH = "match"
    POSSIBLE_MATCH = "possible_match"  # Requires manual review
    NON_MATCH = "non_match"


@dataclass
class MatchField:
    """Configuration for a single matching field in the Fellegi-Sunter model."""
    name: str
    m_probability: float  # P(agree | same person) — typically 0.90-0.99
    u_probability: float  # P(agree | different persons) — depends on field rarity
    missing_weight: float = 0.0  # Weight when field is missing/empty
    comparison_method: str = "exact"  # "exact", "fuzzy", "date", "phonetic"
    weight: float = 0.0  # Computed: log2(m/u) for agree, log2((1-m)/(1-u)) for disagree

    def __post_init__(self):
        if self.u_probability > 0 and self.m_probability > 0:
            self.agree_weight = math.log2(self.m_probability / self.u_probability)
            self.disagree_weight = math.log2(
                (1 - self.m_probability) / (1 - self.u_probability)
            )
        else:
            self.agree_weight = 0.0
            self.disagree_weight = 0.0


@dataclass
class MatchResult:
    """Result of comparing two patient records."""
    record_a_id: str
    record_b_id: str
    composite_score: float
    decision: MatchDecision
    field_scores: dict[str, dict[str, Any]]
    match_threshold: float
    review_threshold: float


@dataclass
class PatientDemographics:
    """Normalized patient demographic record."""
    record_id: str
    first_name: str = ""
    last_name: str = ""
    middle_name: str = ""
    date_of_birth: Optional[date] = None
    sex: str = ""  # M, F, U
    ssn_last_four: str = ""
    phone: str = ""
    address_line: str = ""
    city: str = ""
    state: str = ""
    zip_code: str = ""
    mrn: str = ""
    facility: str = ""


class DemographicNormalizer:
    """
    Normalize patient demographics for matching.

    Handles the messy reality of patient data: nicknames, prefixes,
    suffixes, phone format variations, address abbreviations, etc.
    """

    # Common nickname mappings
    NICKNAMES = {
        "robert": {"bob", "rob", "robbie", "bobby", "bert"},
        "william": {"will", "bill", "billy", "willy", "liam"},
        "james": {"jim", "jimmy", "jamie"},
        "john": {"jack", "johnny", "jon"},
        "richard": {"rick", "dick", "rich", "ricky"},
        "elizabeth": {"liz", "beth", "lizzy", "eliza", "betty", "bess"},
        "margaret": {"maggie", "meg", "peggy", "marge", "margie"},
        "jennifer": {"jen", "jenny", "jenn"},
        "katherine": {"kate", "kathy", "katie", "cathy", "kat"},
        "michael": {"mike", "mikey", "mickey"},
        "joseph": {"joe", "joey"},
        "thomas": {"tom", "tommy"},
        "charles": {"charlie", "chuck", "chas"},
        "patricia": {"pat", "patty", "tricia"},
        "daniel": {"dan", "danny"},
        "christopher": {"chris", "topher"},
        "matthew": {"matt", "matty"},
        "anthony": {"tony", "ant"},
        "david": {"dave", "davey"},
        "edward": {"ed", "eddie", "ted", "teddy", "ned"},
    }

    # Build reverse lookup: nickname -> canonical name
    _NICKNAME_TO_CANONICAL: dict[str, str] = {}
    for canonical, nicks in NICKNAMES.items():
        for nick in nicks:
            _NICKNAME_TO_CANONICAL[nick] = canonical
        _NICKNAME_TO_CANONICAL[canonical] = canonical

    @classmethod
    def normalize_name(cls, name: str) -> str:
        """
        Normalize a name for comparison.
        Lowercase, strip prefixes/suffixes, standardize whitespace.
        """
        if not name:
            return ""
        normalized = name.lower().strip()
        # Remove common prefixes
        for prefix in ("mr.", "mrs.", "ms.", "dr.", "mr ", "mrs ", "ms ", "dr "):
            if normalized.startswith(prefix):
                normalized = normalized[len(prefix):].strip()
        # Remove common suffixes
        for suffix in (" jr", " jr.", " sr", " sr.", " ii", " iii", " iv", " md", " phd"):
            if normalized.endswith(suffix):
                normalized = normalized[:-len(suffix)].strip()
        # Remove non-alpha characters except spaces and hyphens
        normalized = re.sub(r"[^a-z\s\-]", "", normalized)
        # Collapse whitespace
        normalized = re.sub(r"\s+", " ", normalized).strip()
        return normalized

    @classmethod
    def get_canonical_name(cls, name: str) -> str:
        """Map a name or nickname to its canonical form."""
        normalized = cls.normalize_name(name)
        return cls._NICKNAME_TO_CANONICAL.get(normalized, normalized)

    @classmethod
    def normalize_phone(cls, phone: str) -> str:
        """Strip to digits only, keep last 10 (US numbers)."""
        digits = re.sub(r"\D", "", phone)
        if len(digits) == 11 and digits.startswith("1"):
            digits = digits[1:]  # Remove country code
        return digits[-10:] if len(digits) >= 10 else digits

    @classmethod
    def normalize_ssn(cls, ssn: str) -> str:
        """Extract last 4 digits (never store full SSN if you can avoid it)."""
        digits = re.sub(r"\D", "", ssn)
        return digits[-4:] if len(digits) >= 4 else ""

    @classmethod
    def normalize_address(cls, address: str) -> str:
        """Normalize address abbreviations."""
        if not address:
            return ""
        normalized = address.lower().strip()
        abbreviations = {
            "street": "st", "avenue": "ave", "boulevard": "blvd",
            "drive": "dr", "lane": "ln", "road": "rd",
            "court": "ct", "place": "pl", "circle": "cir",
            "apartment": "apt", "suite": "ste", "unit": "unit",
            "north": "n", "south": "s", "east": "e", "west": "w",
        }
        for full, abbr in abbreviations.items():
            normalized = re.sub(rf"\b{full}\b", abbr, normalized)
        # Remove periods, extra whitespace
        normalized = normalized.replace(".", "")
        normalized = re.sub(r"\s+", " ", normalized).strip()
        return normalized

    @classmethod
    def normalize_demographics(cls, demographics: PatientDemographics) -> PatientDemographics:
        """Apply all normalizations to a patient record."""
        return PatientDemographics(
            record_id=demographics.record_id,
            first_name=cls.normalize_name(demographics.first_name),
            last_name=cls.normalize_name(demographics.last_name),
            middle_name=cls.normalize_name(demographics.middle_name),
            date_of_birth=demographics.date_of_birth,
            sex=demographics.sex.upper()[:1] if demographics.sex else "",
            ssn_last_four=cls.normalize_ssn(demographics.ssn_last_four),
            phone=cls.normalize_phone(demographics.phone),
            address_line=cls.normalize_address(demographics.address_line),
            city=cls.normalize_name(demographics.city),
            state=demographics.state.upper()[:2] if demographics.state else "",
            zip_code=re.sub(r"\D", "", demographics.zip_code)[:5],
            mrn=demographics.mrn,
            facility=demographics.facility,
        )


class PatientMatcher:
    """
    Probabilistic patient matcher using the Fellegi-Sunter model.

    The model computes a composite log-likelihood ratio score for each
    record pair based on field-level agreement/disagreement probabilities.

    Match fields are configured with:
    - m-probability: P(field agrees | records are the same person)
    - u-probability: P(field agrees by coincidence | records are different people)

    The ratio m/u for agreement gives higher weight to rare fields
    (SSN agreement is more informative than sex agreement because
    u-probability for SSN is much lower).
    """

    DEFAULT_FIELDS = [
        MatchField("last_name", m_probability=0.95, u_probability=0.01, comparison_method="fuzzy"),
        MatchField("first_name", m_probability=0.92, u_probability=0.02, comparison_method="fuzzy"),
        MatchField("date_of_birth", m_probability=0.97, u_probability=0.0001, comparison_method="date"),
        MatchField("sex", m_probability=0.98, u_probability=0.50, comparison_method="exact"),
        MatchField("ssn_last_four", m_probability=0.99, u_probability=0.0001, comparison_method="exact"),
        MatchField("phone", m_probability=0.90, u_probability=0.001, comparison_method="exact"),
        MatchField("address_line", m_probability=0.85, u_probability=0.01, comparison_method="fuzzy"),
        MatchField("zip_code", m_probability=0.90, u_probability=0.03, comparison_method="exact"),
    ]

    def __init__(
        self,
        match_fields: Optional[list[MatchField]] = None,
        match_threshold: float = 12.0,    # Above this: definite match
        review_threshold: float = 6.0,     # Between review and match: manual review
        normalizer: Optional[DemographicNormalizer] = None,
    ):
        self._fields = match_fields or self.DEFAULT_FIELDS
        self._match_threshold = match_threshold
        self._review_threshold = review_threshold
        self._normalizer = normalizer or DemographicNormalizer()

    def compare(
        self,
        record_a: PatientDemographics,
        record_b: PatientDemographics,
    ) -> MatchResult:
        """
        Compare two patient records using the Fellegi-Sunter model.

        Returns a MatchResult with composite score and per-field breakdown.
        """
        # Normalize both records
        norm_a = self._normalizer.normalize_demographics(record_a)
        norm_b = self._normalizer.normalize_demographics(record_b)

        composite_score = 0.0
        field_scores = {}

        for match_field in self._fields:
            value_a = getattr(norm_a, match_field.name, "")
            value_b = getattr(norm_b, match_field.name, "")

            # Handle missing data
            if not value_a or not value_b:
                field_scores[match_field.name] = {
                    "score": match_field.missing_weight,
                    "comparison": "missing",
                    "value_a": value_a,
                    "value_b": value_b,
                }
                composite_score += match_field.missing_weight
                continue

            # Compare based on method
            agrees = self._compare_field(
                value_a, value_b, match_field.comparison_method, match_field.name
            )

            if agrees:
                score = match_field.agree_weight
                comparison = "agree"
            else:
                score = match_field.disagree_weight
                comparison = "disagree"

            field_scores[match_field.name] = {
                "score": round(score, 4),
                "comparison": comparison,
                "value_a": value_a,
                "value_b": value_b,
            }
            composite_score += score

        # Determine decision
        if composite_score >= self._match_threshold:
            decision = MatchDecision.MATCH
        elif composite_score >= self._review_threshold:
            decision = MatchDecision.POSSIBLE_MATCH
        else:
            decision = MatchDecision.NON_MATCH

        return MatchResult(
            record_a_id=record_a.record_id,
            record_b_id=record_b.record_id,
            composite_score=round(composite_score, 4),
            decision=decision,
            field_scores=field_scores,
            match_threshold=self._match_threshold,
            review_threshold=self._review_threshold,
        )

    def find_duplicates(
        self,
        candidate: PatientDemographics,
        existing_records: list[PatientDemographics],
    ) -> list[MatchResult]:
        """
        Find potential duplicate records for a candidate patient.
        Returns matches and possible matches, sorted by score.
        """
        results = []
        for existing in existing_records:
            result = self.compare(candidate, existing)
            if result.decision in (MatchDecision.MATCH, MatchDecision.POSSIBLE_MATCH):
                results.append(result)

        results.sort(key=lambda r: r.composite_score, reverse=True)
        return results

    def _compare_field(
        self,
        value_a: str,
        value_b: str,
        method: str,
        field_name: str,
    ) -> bool:
        """Compare two field values using the specified method."""
        if method == "exact":
            return value_a == value_b

        elif method == "fuzzy":
            # Use Jaro-Winkler similarity (preferred for names/addresses)
            similarity = self._jaro_winkler_similarity(value_a, value_b)
            # Also check nickname equivalence for name fields
            if field_name == "first_name":
                canonical_a = DemographicNormalizer.get_canonical_name(value_a)
                canonical_b = DemographicNormalizer.get_canonical_name(value_b)
                if canonical_a == canonical_b and canonical_a:
                    return True
            return similarity >= 0.85

        elif method == "date":
            # Date comparison: exact match, or transposition detection
            if value_a == value_b:
                return True
            # Check for common date entry errors (month/day transposition)
            return self._date_transposition_match(value_a, value_b)

        return value_a == value_b

    @staticmethod
    def _jaro_winkler_similarity(s1: str, s2: str, winkler_prefix_weight: float = 0.1) -> float:
        """
        Jaro-Winkler string similarity (0.0 to 1.0).

        Preferred over Levenshtein for name matching because it gives
        higher scores to strings that match from the beginning, which
        is more common in name variations (e.g., "Jon" vs "Jonathan").
        """
        if s1 == s2:
            return 1.0
        if not s1 or not s2:
            return 0.0

        len_s1, len_s2 = len(s1), len(s2)
        match_distance = max(len_s1, len_s2) // 2 - 1
        if match_distance < 0:
            match_distance = 0

        s1_matches = [False] * len_s1
        s2_matches = [False] * len_s2

        matches = 0
        transpositions = 0

        for i in range(len_s1):
            start = max(0, i - match_distance)
            end = min(i + match_distance + 1, len_s2)

            for j in range(start, end):
                if s2_matches[j] or s1[i] != s2[j]:
                    continue
                s1_matches[i] = True
                s2_matches[j] = True
                matches += 1
                break

        if matches == 0:
            return 0.0

        k = 0
        for i in range(len_s1):
            if not s1_matches[i]:
                continue
            while not s2_matches[k]:
                k += 1
            if s1[i] != s2[k]:
                transpositions += 1
            k += 1

        jaro = (
            matches / len_s1
            + matches / len_s2
            + (matches - transpositions / 2) / matches
        ) / 3.0

        # Winkler modification: boost for common prefix
        prefix_len = 0
        for i in range(min(4, min(len_s1, len_s2))):
            if s1[i] == s2[i]:
                prefix_len += 1
            else:
                break

        return jaro + prefix_len * winkler_prefix_weight * (1 - jaro)

    @staticmethod
    def _date_transposition_match(date_a: str, date_b: str) -> bool:
        """
        Detect common date entry errors.

        Month/day transposition (03/12 vs 12/03) is the most common
        date entry error in patient registration. We flag these as
        possible matches requiring review, not definite matches.
        """
        # This is a simplified check; production systems use
        # the actual date objects parsed from the demographic record
        try:
            # Assume date objects stored on PatientDemographics
            # Check if month and day are swapped
            if hasattr(date_a, 'month') and hasattr(date_b, 'month'):
                return (
                    date_a.month == date_b.day
                    and date_a.day == date_b.month
                    and date_a.year == date_b.year
                    and date_a.month <= 12
                    and date_a.day <= 12
                )
        except (AttributeError, TypeError):
            pass
        return False


class MasterPatientIndex:
    """
    Enterprise Master Patient Index (EMPI) operations.

    Manages canonical patient identities across a health system,
    including merge (combining duplicate records) and unmerge
    (splitting incorrectly merged records) operations.
    """

    def __init__(self, storage: Any, audit_logger: Any):
        self._storage = storage
        self._audit = audit_logger
        self._matcher = PatientMatcher()

    def merge_records(
        self,
        surviving_record_id: str,
        deprecated_record_id: str,
        merged_by: str,
        reason: str,
        verification_method: str,
    ) -> dict[str, Any]:
        """
        Merge two patient records into one.

        The surviving record becomes the canonical identity.
        The deprecated record is marked as merged and all references
        are updated to point to the surviving record.

        This operation is AUDITED and REVERSIBLE (via unmerge).
        """
        merge_record = {
            "operation": "merge",
            "merge_id": str(hash(f"{surviving_record_id}:{deprecated_record_id}")),
            "surviving_record_id": surviving_record_id,
            "deprecated_record_id": deprecated_record_id,
            "merged_by": merged_by,
            "reason": reason,
            "verification_method": verification_method,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "status": "completed",
            "affected_systems": [],
            "rollback_available": True,
        }

        # In production:
        # 1. Update all references (orders, results, appointments, encounters)
        #    from deprecated_record_id to surviving_record_id
        # 2. Move unique data elements from deprecated to surviving
        # 3. Mark deprecated record with pointer to surviving
        # 4. Notify downstream systems (lab, pharmacy, radiology)
        # 5. Audit log the entire operation with full before/after state

        return merge_record

    def unmerge_records(
        self,
        merge_id: str,
        unmerged_by: str,
        reason: str,
    ) -> dict[str, Any]:
        """
        Reverse a previous merge operation.

        This is critical when a merge is discovered to be incorrect
        (two different patients were merged into one). The unmerge must:
        1. Restore the deprecated record to active status
        2. Reassign data elements to their original records
        3. Notify downstream systems of the correction
        4. Flag all clinical decisions made during the merge period for review
        """
        unmerge_record = {
            "operation": "unmerge",
            "original_merge_id": merge_id,
            "unmerged_by": unmerged_by,
            "reason": reason,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "status": "completed",
            "clinical_review_required": True,
            "review_note": (
                "All clinical decisions made between the merge and unmerge "
                "timestamps must be reviewed. Orders, medications, allergies, "
                "and diagnoses may have been applied to the wrong patient."
            ),
        }

        return unmerge_record

What AI tools get right: Claude Code generates the Fellegi-Sunter model with correct m-probability and u-probability semantics, and understands why Jaro-Winkler is preferred over Levenshtein for name matching. It also raises the unmerge scenario and the clinical review requirement after an incorrect merge. What they get wrong: Copilot generates simple exact-match SQL queries (WHERE last_name = ? AND first_name = ? AND dob = ?) that miss the entire point of probabilistic matching. Even when prompted for “fuzzy matching,” it uses Levenshtein distance without Jaro-Winkler’s prefix weighting, which is inferior for name matching. No tool except Claude Code handles the nickname equivalence problem (“Robert” should match “Bob”). Windsurf and Tabnine generate generic search/filter patterns that treat patient matching as a simple database lookup. The most dangerous error is that most tools do not implement the date transposition check — month/day swaps (entering 03/12/1990 as 12/03/1990) are the single most common date entry error in patient registration, and missing this match means the system fails to identify a duplicate that a human would immediately recognize.

What AI Tools Get Wrong in Healthcare Code

After testing all six tools across hundreds of healthcare-specific prompts, these are the nine most common and most dangerous errors. Every one of these has real regulatory or patient safety consequences.

1. Logging PHI in error messages and stack traces. This is the single most common HIPAA violation in AI-generated healthcare code. Every tool, including Claude Code, will generate logger.error(f"Failed to process record for patient {patient_name}: {error}") unless you specifically instruct it not to. Application logs are typically stored unencrypted, retained indefinitely, accessible to developers who have no treatment relationship with the patient, and often shipped to third-party log aggregation services (Datadog, Splunk, ELK) that may not have a BAA in place. Every log statement in healthcare code must be reviewed for PHI leakage, and error handling must use patient identifiers that are not themselves PHI (internal UUID, not name or MRN).
2. Not implementing minimum necessary access. AI tools generate data access functions that return entire patient records by default. A function called get_patient_for_billing() should return only the fields a billing specialist needs (name, DOB, MRN, insurance ID, diagnosis codes, procedure codes) — not allergies, medications, clinical notes, or imaging results. The minimum necessary standard (§164.502(b)) is violated every time a function returns more PHI than its stated purpose requires, and AI tools never enforce this unless the access control structure is already in the project context.
3. Using simple string matching for patient names instead of probabilistic matching. Every tool except Claude Code generates WHERE last_name = ? AND first_name = ? for patient lookups. This fails for name variations (“Bob” vs “Robert”), data entry errors (“Smtih” vs “Smith”), cultural name order differences, hyphenated names, and the many other ways real patient names deviate from perfect string equality. The result is duplicate patient records, which cause incomplete medical histories and potential patient safety events.
4. Hardcoding terminology codes instead of using ValueSet bindings. AI tools generate code like if observation.code == "2160-0" (the LOINC code for creatinine) instead of querying against a ValueSet that includes all valid codes for a concept. LOINC codes get updated, concepts have multiple valid codes, and local systems may use different codes for the same measurement. Hardcoded codes break when the upstream system sends a different-but-valid code for the same clinical concept.
5. Missing HL7v2 escape sequences in parsing. HL7v2 uses backslash escape sequences (\F\ for pipe, \S\ for caret, \T\ for ampersand, \R\ for tilde, \E\ for backslash) to represent special characters within field values. Every tool except Claude Code generates HL7v2 parsers that split on pipe characters without first processing escape sequences. This corrupts data when a field value legitimately contains a pipe character — which happens in clinical notes, free-text fields, and Z-segments.
6. Not handling FHIR resource versioning (If-Match/ETag). FHIR servers use optimistic concurrency control via ETag headers. When you read a resource, the server returns an ETag. When you update it, you must send that ETag in an If-Match header. If someone else modified the resource between your read and write, the server returns 409 Conflict. Every AI tool generates FHIR update code without versioning, creating race conditions where concurrent updates silently overwrite each other — losing clinical data like medication changes or allergy additions.
7. Generating drug interaction logic without severity classification. Copilot and Windsurf generate boolean interaction checks (“interaction found: true/false”) without distinguishing between contraindicated combinations (must not co-prescribe) and minor interactions (monitor patient). A system that flags every interaction with equal urgency causes alert fatigue — clinicians override all alerts, including the critical ones. A system that does not flag severity puts the burden on the clinician to look up every interaction manually, which defeats the purpose of CDS.
8. Missing 21 CFR Part 11 electronic signature requirements. AI tools confuse cryptographic digital signatures (RSA, ECDSA) with regulatory electronic signatures. Part 11 electronic signatures require the signer’s printed name, the date and time, the meaning of the signature (review, approval, authorship), two-factor identification verification, and a binding between the signature and the specific record being signed (preventing signature transplantation). These are regulatory requirements, not cryptographic requirements, and no AI tool implements them correctly without explicit prompting.
9. Not encrypting PHI at rest in temporary files and caches. AI tools generate code that writes PHI to temporary files, in-memory caches (Redis), or session storage without encryption. A temporary CSV export of patient lab results, an in-memory cache of recently viewed patient records, a session variable storing the current patient context — all of these contain PHI and must be encrypted at rest per §164.312(a)(2)(iv). The HIPAA encryption addressable implementation specification means you must either encrypt or document why encryption is not reasonable and implement equivalent protections — and “it is a temporary file” is not a valid reason to skip encryption.

Cost Model: Healthcare & Clinical Engineering Scenarios

Healthcare software teams have unique considerations: data governance requirements may prevent sending PHI-adjacent code to cloud AI providers, on-premise deployment may be necessary, and the cost of a regulatory violation dwarfs the subscription cost of any tool. Here are the recommended configurations by team size and regulatory exposure.

A note on data governance: if your development codebase contains PHI, code excerpts that reference real patient identifiers, or configuration files with production database credentials, you must evaluate whether sending that code to a cloud AI provider constitutes a disclosure under HIPAA. Copilot Enterprise ($39/seat) provides the strongest data handling guarantees (SOC 2 Type II, no code retention for training). Claude Code Team ($30/seat) offers zero-retention API usage. Tabnine Enterprise offers on-premise deployment for organizations that cannot send any code to external services. This is not theoretical — HHS OCR has investigated cloud service usage in healthcare and the BAA chain applies to every service that processes PHI, including AI coding assistants if they receive code containing identifiable patient information.

Related Guides

Compare all the tools and pricing on our main comparison table, check the cheapest tools guide for budget options, or see the enterprise guide for organizational procurement and data governance considerations.