Recipe 5.8: Python Implementation Example

pip install boto3

import hashlib
import hmac
import json
import logging
import math
import re
import secrets
import unicodedata
import uuid
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional

import boto3
from boto3.dynamodb.conditions import Key
from botocore.config import Config

# Structured logging. In production, ship JSON-formatted records
# to CloudWatch Logs Insights. Encoded records are not PHI by
# themselves (the cryptographic transform is the privacy claim),
# but the per-record consent posture, the cohort-axis hashes, and
# the source-record identifiers (retained locally at each
# participant) all carry information that should not leak through
# logs. Log structural metadata only (cycle_id, encoded_record_id,
# parameterization_version, salt_key_version, decision band),
# never CLK payloads, never raw demographics, never source-record
# identifiers from cross-participant context.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling from DynamoDB, EventBridge,
# CloudWatch, S3, and SQS. Per-cycle linkage latencies are not
# real-time critical (research collaborations and public-health
# surveillance run on cycle cadences of weekly to quarterly), but
# bulk encoding runs are throughput-sensitive enough that
# transient throttling on any one service should not fail an
# entire Glue job. Step Functions Catch states distinguish
# retriable infrastructure failures from terminal logic failures
# and route terminal failures to a DLQ for human investigation.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 5, "mode": "adaptive"})

# Module-level clients. Reused across Lambda invocations in warm
# containers so each invocation does not pay the connection cost.
REGION = "us-east-1"
dynamodb           = boto3.resource("dynamodb", region_name=REGION, config=BOTO3_RETRY_CONFIG)
s3_client          = boto3.client("s3",          region_name=REGION, config=BOTO3_RETRY_CONFIG)
sqs_client         = boto3.client("sqs",         region_name=REGION, config=BOTO3_RETRY_CONFIG)
eventbridge_client = boto3.client("events",      region_name=REGION, config=BOTO3_RETRY_CONFIG)
cloudwatch_client  = boto3.client("cloudwatch",  region_name=REGION, config=BOTO3_RETRY_CONFIG)
kms_client         = boto3.client("kms",         region_name=REGION, config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Fill these in with your actual resource names. The demo prints
# what it would write rather than failing if the resources do not
# exist; see run_demo() at the bottom.
LINKAGE_CYCLE_TABLE         = "linkage-cycle-metadata"
LINKAGE_RESULTS_TABLE       = "linkage-results-store"
PROTOCOL_CONFIG_TABLE       = "protocol-parameterization-config"
PARTICIPANT_LOCAL_MAP_TABLE = "participant-local-mapping"
PARTICIPANT_A_BUCKET        = "participant-a-encoded-records"
PARTICIPANT_B_BUCKET        = "participant-b-encoded-records"
AUDIT_ARCHIVE_BUCKET        = "pprl-audit-archive"
DERIVED_SNAPSHOT_BUCKET     = "pprl-derived-snapshots"
LINKAGE_REVIEW_QUEUE_URL    = "https://sqs.us-east-1.amazonaws.com/000000000000/pprl-linkage-review-queue"
SALT_ROTATION_QUEUE_URL     = "https://sqs.us-east-1.amazonaws.com/000000000000/pprl-salt-rotation-queue"
CONSENT_WITHDRAWAL_QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/000000000000/pprl-consent-withdrawal-queue"
PROPAGATION_QUEUE_URL       = "https://sqs.us-east-1.amazonaws.com/000000000000/pprl-propagation-queue"
PPRL_EVENT_BUS_NAME         = "pprl-events-bus"
CLOUDWATCH_NAMESPACE        = "PrivacyPreservingRecordLinkage/Cycle"

# Deploy-time guardrail. Any blank resource name is a deploy-time
# bug, not a runtime surprise.
for _name, _value in [
    ("LINKAGE_CYCLE_TABLE",         LINKAGE_CYCLE_TABLE),
    ("LINKAGE_RESULTS_TABLE",       LINKAGE_RESULTS_TABLE),
    ("PROTOCOL_CONFIG_TABLE",       PROTOCOL_CONFIG_TABLE),
    ("PARTICIPANT_LOCAL_MAP_TABLE", PARTICIPANT_LOCAL_MAP_TABLE),
    ("PARTICIPANT_A_BUCKET",        PARTICIPANT_A_BUCKET),
    ("PARTICIPANT_B_BUCKET",        PARTICIPANT_B_BUCKET),
    ("AUDIT_ARCHIVE_BUCKET",        AUDIT_ARCHIVE_BUCKET),
    ("DERIVED_SNAPSHOT_BUCKET",     DERIVED_SNAPSHOT_BUCKET),
    ("LINKAGE_REVIEW_QUEUE_URL",    LINKAGE_REVIEW_QUEUE_URL),
    ("SALT_ROTATION_QUEUE_URL",     SALT_ROTATION_QUEUE_URL),
    ("CONSENT_WITHDRAWAL_QUEUE_URL", CONSENT_WITHDRAWAL_QUEUE_URL),
    ("PROPAGATION_QUEUE_URL",       PROPAGATION_QUEUE_URL),
    ("PPRL_EVENT_BUS_NAME",         PPRL_EVENT_BUS_NAME),
    ("CLOUDWATCH_NAMESPACE",        CLOUDWATCH_NAMESPACE),
]:
    assert _value, f"{_name} must be set before deploying."

# --- Versioning ---
# Every encoded record and every linkage-cycle metadata row
# carries the parameterization version and the salt-key version
# active at decision time. This is how a future audit
# reconstructs which cryptographic primitives were active when a
# particular linkage was made, and how the invalidation pipeline
# knows which records to re-encode when the salt rotates.
PROTOCOL_PARAMETERIZATION_VERSION = "pprl-clk-v2.3.1"
MATCHER_CONFIG_VERSION            = "pprl-fs-v1.4.2"

# --- Protocol parameterization (CLK Bloom-filter encoding) ---
# These parameters control the cryptographic encoding. They are
# negotiated cross-participant before any encoding happens and
# pinned per cycle. Mis-coordinated parameters (one party uses
# n=2, the other uses n=3; one party uses k=30, the other uses
# k=20; one party uses a 1024-bit filter, the other uses 2048)
# produce encoded records that are not comparable, and the
# matcher silently returns zero matches.
PROTOCOL_PARAMETERIZATION = {
    "parameterization_version": PROTOCOL_PARAMETERIZATION_VERSION,
    "n_gram_size":              2,       # bigrams
    "bloom_filter_size":        1024,    # total bits
    "hash_function_count":      30,      # k value
    "per_feature_bit_allocation": {
        # Each feature gets a share of the total filter. Names
        # get more bits because they're the most discriminating
        # feature; DOB and SSN-last-4 are tokenized (set-bit-
        # exact) rather than n-gram-encoded so they don't need
        # a wide allocation.
        "given_name":   200,
        "middle_name":  100,
        "family_name":  300,
        "dob":          150,
        "address_line": 150,
        "zip_code":     50,
        "phone":        50,
        "ssn_last_4":   24,
    },
    "defensive_measures": {
        # Production CLK encoders apply random hashing (varies
        # the hash-function-to-bit-position mapping per record),
        # balanced encoding (ensures each filter has a similar
        # number of set bits to defeat frequency analysis), and
        # hardening (deliberate noise injection that defeats
        # specific known attacks). The demo's defensive measures
        # are deliberately simple; production uses clkhash which
        # implements current-best-practice variants.
        "random_hashing_enabled":   False,
        "balanced_encoding_enabled": False,
        "hardening_enabled":         False,
    },
}

# --- Encoded-data thresholds ---
# Calibrated SEPARATELY from the conventional matcher's
# thresholds in recipe 5.1 / 5.5. The encoded-data scoring
# function is different (Sørensen-Dice over Bloom filters
# combined Fellegi-Sunter-style) and the absolute scale of
# similarity scores is different. Re-using the recipe 5.5
# thresholds for PPRL produces silent linkage failures.
# Calibrate against a known-overlap pilot population encoded
# under the production parameterization, with input from the
# institution's analytics governance committee, the privacy
# team, and the cohort-equity-monitoring committee.
ENCODED_MATCH_HIGH_THRESHOLD = Decimal("0.85")
ENCODED_MATCH_MED_THRESHOLD  = Decimal("0.72")
ENCODED_REJECT_THRESHOLD     = Decimal("0.50")

# --- Per-feature weights for the Fellegi-Sunter combiner ---
# Same probabilistic-record-linkage core as recipes 5.1 / 5.5.
# The weights reflect the relative discriminating power of each
# feature plus the encoded-data noise level (more noise = less
# weight). DOB carries less weight here than in plaintext
# matching because the encoded-data matcher is comparing token-
# level set-membership rather than exact string match, so the
# feature is more vulnerable to encoding noise.
FEATURE_WEIGHTS = {
    "given_name":   Decimal("0.18"),
    "middle_name":  Decimal("0.08"),
    "family_name":  Decimal("0.22"),
    "dob":          Decimal("0.20"),
    "address_line": Decimal("0.10"),
    "zip_code":     Decimal("0.05"),
    "phone":        Decimal("0.07"),
    "ssn_last_4":   Decimal("0.10"),
}

# --- Disclosure forms ---
# Each form has its own privacy properties. The protocol's trust
# framework specifies which form(s) a particular linkage is
# authorized to produce. Some research collaborations authorize
# only intersection_count; others authorize per_record_match_flags
# for each participant restricted to its own records; others
# authorize differentially_private_aggregate for cross-cohort
# analytic queries.
DISCLOSURE_FORMS = {
    "per_record_match_flags",          # per-record yes/no flags
    "intersection_count",              # size only
    "k_anonymous_aggregate",           # cohort aggregates with suppression
    "differentially_private_aggregate", # aggregates with DP noise
    "encrypted_match_indicator",       # encrypted under consumer key
}

# --- Default k-anonymity and differential-privacy parameters ---
# The disclosure-policy specifies these per use case. Values
# below are illustrative defaults; institutional privacy team
# owns the per-use-case parameters.
DEFAULT_K_ANONYMITY        = 11
DEFAULT_SUPPRESSION_THRESHOLD = 5
DEFAULT_DP_EPSILON         = Decimal("1.0")
DEFAULT_DP_DELTA           = Decimal("0.000001")  # 1e-6

# --- Cohort axes ---
# Each participant computes its own cohort-axis values locally
# (using its own demographic visibility) and contributes the
# *hashes* of those values in the encoded-record envelope. The
# matcher receives the hashes (not the values) and can stratify
# accuracy metrics by hash without learning the underlying axis
# values. Production includes additional axes specific to the
# institution's equity-monitoring program; the demo's three are
# illustrative.
COHORT_AXES = ["name_tradition", "age_decade", "sex_or_gender"]

def _to_decimal(value) -> Decimal:
    """Coerce numeric input into Decimal for DynamoDB."""
    if isinstance(value, Decimal):
        return value
    return Decimal(str(value))

def _now_iso() -> str:
    """UTC timestamp in ISO 8601 format. Always UTC; never local time."""
    return datetime.now(timezone.utc).isoformat()

def _strip_diacritics(s: str) -> str:
    """Strip combining diacritical marks for cross-participant
    canonicalization. Critical for names with accents that one
    participant's EHR may have stripped on input while another
    participant's preserved them. If the canonicalization differs
    cross-participant, the encoded records will not match even
    when they refer to the same person."""
    if not s:
        return ""
    nfkd = unicodedata.normalize("NFKD", s)
    return "".join(c for c in nfkd if not unicodedata.combining(c))

def _canonical_name(*parts) -> str:
    """Normalize a name to canonical lowercase whitespace-collapsed
    form for encoding. Production handles per-tradition rules
    (Spanish double-surname order, East Asian family-name-first
    conventions, Arabic patronymic structures) here too. The
    canonicalization spec is part of the protocol parameterization
    and is the same across all participants."""
    joined = " ".join(str(p or "").strip() for p in parts)
    joined = _strip_diacritics(joined).lower()
    joined = re.sub(r"[^\w\s'-]", " ", joined)
    joined = re.sub(r"\s+", " ", joined).strip()
    return joined

def _produce_n_grams(value: str, n: int = 2,
                       include_markers: bool = True) -> list:
    """Tokenize a string into n-grams. Underscores serve as
    start-and-end markers that improve robustness on short
    strings (the underscore counts as the start-of-string token
    so 'Bob' becomes ['_B', 'Bo', 'ob', 'b_'] rather than just
    ['Bo', 'ob'])."""
    if not value:
        return []
    if include_markers:
        value = "_" + value + "_"
    return [value[i:i + n] for i in range(len(value) - n + 1)]

def _compute_per_feature_size(parameterization: dict,
                                feature_name: str) -> int:
    """Look up the feature's bit allocation from the protocol
    parameterization."""
    return parameterization["per_feature_bit_allocation"].get(
        feature_name, 0)

def _hmac_sha_256(key: bytes, message: bytes) -> bytes:
    """HMAC-SHA-256 keyed hash. The salt is the HMAC key; varying
    the key per hash-function-index gives k independent hash
    functions. Production uses HSM-backed HMAC where the key
    plaintext never leaves the hardware-security-module; the
    demo uses a simple hmac.new() for readability."""
    return hmac.new(key, message, hashlib.sha256).digest()

def _hash_to_bit_position(salt: bytes, k_index: int, n_gram: str,
                              modulus: int) -> int:
    """Produce a bit position by HMAC'ing the n-gram under a
    salt-derived key for the given hash-function index. Modulo
    the bit array size to land in range."""
    # Derive the per-k-index key from the salt by HMAC'ing the
    # k_index. Production uses a more robust key-derivation
    # scheme (HKDF with explicit info parameter); the demo's
    # simple HMAC suffices for illustration.
    per_k_key = _hmac_sha_256(salt, k_index.to_bytes(4, "big"))
    h = _hmac_sha_256(per_k_key, n_gram.encode("utf-8"))
    # Take first 8 bytes as a 64-bit integer; mod into the bit array.
    return int.from_bytes(h[:8], "big") % modulus

def _bit_array(size: int) -> bytearray:
    """Initialize an empty bit array of the given size in bits."""
    return bytearray((size + 7) // 8)

def _set_bit(bits: bytearray, position: int) -> None:
    """Set the bit at the given position in the bit array."""
    bits[position // 8] |= (1 << (position % 8))

def _is_bit_set(bits: bytearray, position: int) -> bool:
    """Test whether the bit at the given position is set."""
    return bool(bits[position // 8] & (1 << (position % 8)))

def _count_set_bits(bits: bytearray) -> int:
    """Population count over the bit array. Used for the
    Sørensen-Dice denominator and for balanced-encoding
    diagnostics."""
    return sum(bin(b).count("1") for b in bits)

def _bitwise_and_count(bits_a: bytearray, bits_b: bytearray) -> int:
    """Count of bits set in the bitwise AND of two filters; the
    Sørensen-Dice numerator's intersection size."""
    if len(bits_a) != len(bits_b):
        raise ValueError(
            "Bloom filter size mismatch (parameterization "
            "mis-coordination). This is the classic silent-failure "
            "mode for PPRL: encoded records produced under "
            "different parameterization versions are not "
            "comparable. The cycle's parameterization version "
            "should have been validated at exchange time.")
    return sum(bin(a & b).count("1") for a, b in zip(bits_a, bits_b))

def _sorensen_dice_coefficient(bits_a: bytearray,
                                  bits_b: bytearray) -> Decimal:
    """Sørensen-Dice over two Bloom filters: 2 * |A AND B| /
    (|A| + |B|). Equivalent to F1 of bit-set-membership across
    the two filters; ranges 0.0 (no shared bits) to 1.0
    (identical filters)."""
    set_a = _count_set_bits(bits_a)
    set_b = _count_set_bits(bits_b)
    if set_a + set_b == 0:
        return Decimal("0")
    intersection = _bitwise_and_count(bits_a, bits_b)
    return Decimal("2") * Decimal(intersection) / Decimal(set_a + set_b)

def _sha256(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()

def _serialize_for_dynamodb(obj):
    """Recursive serialization helper. Same pattern as recipes
    5.1 - 5.7."""
    if isinstance(obj, dict):
        return {k: _serialize_for_dynamodb(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_serialize_for_dynamodb(v) for v in obj]
    if isinstance(obj, set):
        return [_serialize_for_dynamodb(v) for v in sorted(obj)]
    if isinstance(obj, (bytes, bytearray)):
        # Bit arrays serialize to base64 for DynamoDB / S3.
        import base64
        return base64.b64encode(bytes(obj)).decode("ascii")
    if isinstance(obj, float):
        return Decimal(str(obj))
    return obj

def _emit_metric(metric_name: str, value: float,
                  dimensions: dict = None) -> None:
    """CloudWatch metric emit. Cohort-bucket-hash dimensions feed
    the cohort-stratified accuracy monitoring; production
    aggregates by CohortBucketHash and alarms on per-cohort
    linkage-rate or false-acceptance-rate disparities."""
    try:
        cloudwatch_client.put_metric_data(
            Namespace=CLOUDWATCH_NAMESPACE,
            MetricData=[{
                "MetricName": metric_name,
                "Value": value,
                "Unit": "Count",
                "Dimensions": [
                    {"Name": k, "Value": v} for k, v in (dimensions or {}).items()
                ],
            }],
        )
    except Exception as exc:
        logger.warning("metric emit failed",
                        extra={"metric": metric_name, "error": str(exc)})

def _archive_to_s3(payload: dict, bucket: str, partition: str,
                     key_id: str = None) -> None:
    """Best-effort archive to S3 with KMS encryption. Failures
    logged and skipped (the demo prints what it would write)."""
    today = datetime.now(timezone.utc).strftime("%Y/%m/%d")
    kid = key_id or uuid.uuid4().hex
    key = f"{partition}/{today}/{kid}.json"
    body = json.dumps(payload, default=str).encode("utf-8")
    try:
        s3_client.put_object(
            Bucket=bucket, Key=key, Body=body,
            ServerSideEncryption="aws:kms",
        )
    except Exception as exc:
        logger.info("archive write skipped (demo mode is fine to ignore)",
                     extra={"bucket": bucket, "key": key, "error": str(exc)})

# --- Mock salt custody ---
class MockSaltCustody:
    """Stand-in for AWS CloudHSM (high-assurance) or KMS with
    customer-managed keys (lower-assurance) salt custody. The
    salt is the cryptographic root-of-trust for the protocol;
    the plaintext salt is never exposed to the encoding process
    in production (HMAC operations call into the HSM context).
    The demo uses an in-memory salt for readability; a production
    deployment would never have plaintext salt visible to Python
    application code."""

    def __init__(self):
        # Per-cycle salt-key versions. The active cycle uses the
        # most recent rotation. Prior versions are retained for
        # the catch-up window (where the prior salt remains valid
        # for read-only operations during a coordinated rotation
        # cutover) and then decommissioned.
        self._salt_keys = {
            "salt-2026-q2-rotation-001": secrets.token_bytes(32),
        }
        self._active_version = "salt-2026-q2-rotation-001"
        self._access_log: list = []

    def get_salt(self, salt_key_version: str,
                  caller_role: str, cycle_id: str) -> bytes:
        """Acquire the salt for the given version. Production's
        equivalent never returns plaintext; it returns an HSM
        context that the encoder calls into for HMAC operations."""
        # Dual-control approval would be enforced here in
        # production; the demo logs but does not enforce.
        self._access_log.append({
            "salt_key_version": salt_key_version,
            "caller_role":      caller_role,
            "cycle_id":         cycle_id,
            "operation":        "get_salt",
            "timestamp":        _now_iso(),
        })
        if salt_key_version not in self._salt_keys:
            raise KeyError(
                f"salt_key_version {salt_key_version} not found "
                f"or expired (the cycle's parameterization "
                f"version may reference a salt that has been "
                f"decommissioned; re-encode under the current "
                f"salt and re-run the cycle)")
        return self._salt_keys[salt_key_version]

    def rotate_salt(self, new_version: str,
                     dual_control_approvers: list) -> None:
        """Rotation ceremony. Production requires dual-control
        approval (two HSM operators from non-overlapping
        organizations); rotation is audit-logged with both
        operator identities."""
        if len(dual_control_approvers) < 2:
            raise ValueError(
                "salt rotation requires dual-control approval "
                "(at least two approvers from non-overlapping "
                "organizations)")
        self._salt_keys[new_version] = secrets.token_bytes(32)
        prior_active = self._active_version
        self._active_version = new_version
        self._access_log.append({
            "operation":        "rotate_salt",
            "new_version":      new_version,
            "prior_active":     prior_active,
            "approvers":        dual_control_approvers,
            "timestamp":        _now_iso(),
        })
        # Production keeps the prior salt available for the
        # catch-up window, then decommissions; the demo retains
        # all versions for inspection.

    def get_active_version(self) -> str:
        return self._active_version

    def get_access_log(self) -> list:
        return list(self._access_log)

# --- Mock protocol parameterization store ---
class MockProtocolParameterizationStore:
    """Stand-in for the versioned configuration store. Production
    reads from a DynamoDB table maintained by the multi-party
    trust framework's governance committee. Each linkage cycle
    pins the parameterization version active at the cycle's start
    and references that version throughout. Updates require
    governance approval and trigger coordinated re-encoding."""

    def __init__(self):
        self._configs = {
            PROTOCOL_PARAMETERIZATION_VERSION: dict(PROTOCOL_PARAMETERIZATION),
        }

    def load_parameterization(self, version: str) -> dict:
        if version not in self._configs:
            raise KeyError(
                f"parameterization version {version} not found")
        return dict(self._configs[version])

    def publish_new_version(self, version: str,
                              parameterization: dict) -> None:
        """A new parameterization version goes through governance
        approval before publish. The demo just stores it."""
        self._configs[version] = dict(parameterization)

# --- Mock participant data sources ---
# Each participant's source data is a list of records with
# standardized demographic features, per-record consent posture,
# and (cohort-axis values used for the cohort-axis-hash
# computation). The Catherine / Maria / Margaret patients
# overlap across participants so the matcher has something to
# find; a few non-overlapping records exercise the no-match path.
SYNTHETIC_PARTICIPANT_A_RECORDS = [
    {
        "source_record_id": "academic-mc-mrn-00284271",
        "given_name":   "Catherine",
        "middle_name":  "Marie",
        "family_name":  "Hernandez",
        "dob":          "1985-03-22",
        "address_line": "412 Oak St",
        "zip_code":     "62701",
        "phone":        "+15551234567",
        "ssn_last_4":   "4287",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           "post_dobbs_state_overlay_v1",
            "consent_recorded_at":                      "2026-01-15T10:34:22Z",
        },
        "cohort_values": {
            "name_tradition":  "spanish_double_surname",
            "age_decade":      "40s",
            "sex_or_gender":   "F",
        },
    },
    {
        "source_record_id": "academic-mc-mrn-00891344",
        "given_name":   "Maria",
        "middle_name":  None,
        "family_name":  "Garcia-Lopez",
        "dob":          "1972-03-14",
        "address_line": "8810 Maple Ave",
        "zip_code":     "60618",
        "phone":        "+15557654321",
        "ssn_last_4":   "1119",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-11-04T08:12:11Z",
        },
        "cohort_values": {
            "name_tradition":  "spanish_double_surname",
            "age_decade":      "50s",
            "sex_or_gender":   "F",
        },
    },
    {
        "source_record_id": "academic-mc-mrn-00733215",
        "given_name":   "Margaret",
        "middle_name":  None,
        "family_name":  "Chen-Patel",
        "dob":          "1990-11-04",
        "address_line": "27 Beacon Hill",
        "zip_code":     "02108",
        "phone":        "+15558889999",
        "ssn_last_4":   "9023",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-10-22T14:55:09Z",
        },
        "cohort_values": {
            "name_tradition":  "east_asian_traditional",
            "age_decade":      "30s",
            "sex_or_gender":   "F",
        },
    },
    {
        "source_record_id": "academic-mc-mrn-00499812",
        "given_name":   "Theodore",
        "middle_name":  "James",
        "family_name":  "Williamson",
        "dob":          "1956-07-29",
        "address_line": "5532 Birch Ln",
        "zip_code":     "02115",
        "phone":        "+15554443333",
        "ssn_last_4":   "2244",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-09-02T11:20:00Z",
        },
        "cohort_values": {
            "name_tradition":  "english_traditional",
            "age_decade":      "60s",
            "sex_or_gender":   "M",
        },
    },
    {
        "source_record_id": "academic-mc-mrn-00622441",
        # This record has consent withdrawn for research linkage;
        # the encoding step filters it out at standardize time.
        "given_name":   "Patricia",
        "middle_name":  "Anne",
        "family_name":  "Murphy",
        "dob":          "1968-12-15",
        "address_line": "204 Spruce St",
        "zip_code":     "62704",
        "phone":        "+15552223333",
        "ssn_last_4":   "8801",
        "consent_posture": {
            "consent_for_research_linkage":             False,  # withdrawn
            "consent_for_outcomes_research_purpose":    False,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-12-01T16:30:00Z",
        },
        "cohort_values": {
            "name_tradition":  "irish_traditional",
            "age_decade":      "50s",
            "sex_or_gender":   "F",
        },
    },
]

SYNTHETIC_PARTICIPANT_B_RECORDS = [
    {
        # Catherine: matches Participant A's first record. Slight
        # address variation (St. vs Street) tests the encoding's
        # robustness to legitimate-noise variation.
        "source_record_id": "regional-payer-member-MEM-100874-A",
        "given_name":   "Catherine",
        "middle_name":  "Marie",
        "family_name":  "Hernandez",
        "dob":          "1985-03-22",
        "address_line": "412 Oak Street",  # "Street" vs "St"
        "zip_code":     "62701",
        "phone":        "+15551234567",
        "ssn_last_4":   "4287",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           "post_dobbs_state_overlay_v1",
            "consent_recorded_at":                      "2026-01-22T09:18:44Z",
        },
        "cohort_values": {
            "name_tradition":  "spanish_double_surname",
            "age_decade":      "40s",
            "sex_or_gender":   "F",
        },
    },
    {
        # Maria: matches Participant A's second record exactly.
        "source_record_id": "regional-payer-member-MEM-303441-C",
        "given_name":   "Maria",
        "middle_name":  None,
        "family_name":  "Garcia-Lopez",
        "dob":          "1972-03-14",
        "address_line": "8810 Maple Ave",
        "zip_code":     "60618",
        "phone":        "+15557654321",
        "ssn_last_4":   "1119",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-11-15T14:00:00Z",
        },
        "cohort_values": {
            "name_tradition":  "spanish_double_surname",
            "age_decade":      "50s",
            "sex_or_gender":   "F",
        },
    },
    {
        # Margaret: matches Participant A's third record, with a
        # phone-number variation that tests the matcher's
        # robustness to feature-level noise.
        "source_record_id": "regional-payer-member-MEM-440412-D",
        "given_name":   "Margaret",
        "middle_name":  None,
        "family_name":  "Chen-Patel",
        "dob":          "1990-11-04",
        "address_line": "27 Beacon Hill",
        "zip_code":     "02108",
        "phone":        "+15558880000",  # phone differs
        "ssn_last_4":   "9023",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-11-08T10:45:30Z",
        },
        "cohort_values": {
            "name_tradition":  "east_asian_traditional",
            "age_decade":      "30s",
            "sex_or_gender":   "F",
        },
    },
    {
        # Non-overlapping member: no match in Participant A.
        "source_record_id": "regional-payer-member-MEM-521098-X",
        "given_name":   "Robert",
        "middle_name":  None,
        "family_name":  "Andersen",
        "dob":          "1948-05-30",
        "address_line": "11 Pine St",
        "zip_code":     "10025",
        "phone":        "+15556667777",
        "ssn_last_4":   "0066",
        "consent_posture": {
            "consent_for_research_linkage":             True,
            "consent_for_outcomes_research_purpose":    True,
            "jurisdictional_overlay_applied":           None,
            "consent_recorded_at":                      "2025-08-15T13:00:00Z",
        },
        "cohort_values": {
            "name_tradition":  "scandinavian_traditional",
            "age_decade":      "70s",
            "sex_or_gender":   "M",
        },
    },
]

# --- Mock consent and jurisdictional-overlay store ---
class MockConsentStore:
    """Stand-in for the institutional consent-management workflow.
    Per-record consent posture is captured at intake and updated
    over time; the encoding step's standardize-and-prepare phase
    consults this store to filter out records the patient has
    not consented to include in this specific linkage."""

    def __init__(self):
        self._withdrawals: dict = {}  # source_record_id -> withdrawn_at

    def is_consent_active(self, record: dict, purpose: str) -> bool:
        """Return whether the record's consent posture permits
        inclusion for the specified purpose. The demo's logic is
        simple; production has a structured policy engine."""
        sid = record.get("source_record_id")
        if sid in self._withdrawals:
            return False
        consent = record.get("consent_posture") or {}
        if purpose == "research_linkage":
            return bool(consent.get("consent_for_research_linkage"))
        if purpose == "outcomes_research":
            return bool(consent.get("consent_for_outcomes_research_purpose"))
        return False

    def withdraw_consent(self, source_record_id: str) -> None:
        """Patient withdraws consent. Forward-only: future cycles
        exclude the record; prior cycles' results remain in the
        consumer's possession."""
        self._withdrawals[source_record_id] = _now_iso()

class MockJurisdictionalOverlays:
    """Stand-in for the institutional policy-overlay store.
    Real deployments encode state law (post-Dobbs reproductive-
    health-care state laws, 42 CFR Part 2 substance-use-treatment
    rules, gender-affirming-care state-law overlays, HIV-and-
    genetic-information rules), institutional policy, and trust-
    framework constraints with attorney-reviewed rules."""

    def applicable_overlays(self, record: dict, purpose: str) -> list:
        overlays = []
        consent = record.get("consent_posture") or {}
        applied = consent.get("jurisdictional_overlay_applied")
        if applied:
            overlays.append({
                "overlay_id":  applied,
                "applies_to":  purpose,
                "constraints": {
                    "audit_every_disclosure": True,
                    "audit_every_query":      True,
                },
            })
        return overlays

# --- Module-level singletons for the demo ---
salt_custody                 = MockSaltCustody()
protocol_config_store        = MockProtocolParameterizationStore()
consent_store                = MockConsentStore()
jurisdictional_overlays_db   = MockJurisdictionalOverlays()
_IN_MEMORY_OUTBOX:           list = []
_IN_MEMORY_REVIEW_QUEUE:     list = []
_IN_MEMORY_LINKAGE_RESULTS:  list = []
_IN_MEMORY_PARTICIPANT_LOCAL_MAPPINGS: dict = {}
# Cohort-axis hash key. Production loads this from Secrets
# Manager and rotates separately from the salt; the demo uses a
# fixed value for reproducibility.
_COHORT_AXIS_HASH_KEY = b"demo-cohort-axis-hash-key-not-for-production"

def _normalize_address(address: str) -> str:
    """USPS-style standardization (light version). Production uses
    a real USPS-CASS-certified standardizer; the demo handles a
    few common variants. The address standardization spec is part
    of the protocol parameterization and is identical across all
    participants."""
    if not address:
        return ""
    s = address.strip()
    # Common abbreviation expansions (standardize TO the
    # abbreviated form so participants converge on the same
    # representation; production picks one canonical form).
    replacements = [
        (r"\bStreet\b",  "St"),
        (r"\bAvenue\b",  "Ave"),
        (r"\bBoulevard\b", "Blvd"),
        (r"\bDrive\b",   "Dr"),
        (r"\bLane\b",    "Ln"),
        (r"\bRoad\b",    "Rd"),
        (r"\bCourt\b",   "Ct"),
        (r"\bApartment\b", "Apt"),
        (r"\bSuite\b",   "Ste"),
    ]
    for pattern, repl in replacements:
        s = re.sub(pattern, repl, s, flags=re.IGNORECASE)
    return _canonical_name(s)

def _normalize_phone(phone: str) -> str:
    """Strip non-digit characters and (for the demo) keep the
    last 10 digits. Production handles country codes, extensions,
    and the e164 canonical form."""
    if not phone:
        return ""
    digits = re.sub(r"\D", "", phone)
    return digits[-10:] if len(digits) >= 10 else digits

def _compute_cohort_axis_hashes(record: dict,
                                  cohort_axes: list,
                                  hash_key: bytes) -> dict:
    """Each participant computes its own cohort-axis values
    locally and contributes the hashes (not the values) in the
    encoded payload. The matcher receives the hashes and can
    stratify accuracy metrics by hash without learning the
    underlying axis values."""
    cohort_values = record.get("cohort_values") or {}
    hashes = {}
    for axis in cohort_axes:
        value = cohort_values.get(axis, "unknown")
        # HMAC the axis-name + value under the hash key so the
        # cohort-axis-hash space is per-axis disjoint (a
        # 'name_tradition' hash cannot collide with a 'sex_or_gender'
        # hash for the same value).
        h = _hmac_sha_256(hash_key,
                            f"{axis}:{value}".encode("utf-8"))
        hashes[f"{axis}_hash"] = h.hex()[:16]
    return hashes

def standardize_and_prepare(source_record_batch: list,
                              participant_id: str,
                              purpose: str,
                              parameterization: dict) -> list:
    """
    Apply the protocol's shared schema to each source record.
    Filter out records whose consent posture does not permit
    inclusion. Compute cohort-axis hashes locally. Return the
    standardized batch ready for encoding.
    """
    standardized = []
    filtered_count = 0

    for source_record in source_record_batch:
        # 1A: apply the consent and purpose-of-use filter.
        if not consent_store.is_consent_active(source_record, purpose):
            filtered_count += 1
            continue

        # 1B: standardize each demographic feature under the
        # protocol's shared schema. Every participant runs this
        # exact same canonicalization; any divergence produces
        # encoded records that the matcher cannot reliably
        # compare.
        normalized = {
            "given_name":   _canonical_name(
                source_record.get("given_name")),
            "middle_name":  _canonical_name(
                source_record.get("middle_name") or ""),
            "family_name":  _canonical_name(
                source_record.get("family_name")),
            "dob":          (source_record.get("dob") or "").strip(),
            "address_line": _normalize_address(
                source_record.get("address_line") or ""),
            "zip_code":     re.sub(
                r"\D", "",
                source_record.get("zip_code") or "")[:5],
            "phone":        _normalize_phone(
                source_record.get("phone") or ""),
            "ssn_last_4":   re.sub(
                r"\D", "",
                source_record.get("ssn_last_4") or "")[-4:],
        }

        # 1C: compute cohort-axis hashes locally.
        cohort_axis_hashes = _compute_cohort_axis_hashes(
            source_record, COHORT_AXES, _COHORT_AXIS_HASH_KEY)

        # 1D: tag with metadata. The source_record_id is retained
        # locally at the participant; it is NOT included in the
        # encoded payload that gets exchanged.
        prepared_record = {
            "participant_id":      participant_id,
            "source_record_id":    source_record["source_record_id"],
            "normalized_features": normalized,
            "consent_posture":     dict(source_record.get(
                                        "consent_posture") or {}),
            "cohort_axis_hashes":  cohort_axis_hashes,
            "applicable_overlays": jurisdictional_overlays_db
                                     .applicable_overlays(
                                         source_record, purpose),
            "prepared_at":         _now_iso(),
        }
        standardized.append(prepared_record)

    _emit_metric("StandardizeAndPrepare.Filtered",
                  float(filtered_count),
                  dimensions={"ParticipantId": participant_id,
                                "Purpose":      purpose})
    _emit_metric("StandardizeAndPrepare.Prepared",
                  float(len(standardized)),
                  dimensions={"ParticipantId": participant_id,
                                "Purpose":      purpose})
    return standardized

def _encode_per_feature_bloom_filter(value: str,
                                          feature_name: str,
                                          parameterization: dict,
                                          salt: bytes) -> bytearray:
    """Produce a per-feature Bloom filter for a single
    demographic feature value. The per-feature filter size is
    derived from the parameterization's per-feature bit
    allocation; n-grams are tokenized; each n-gram is hashed by
    k cryptographic hash functions parameterized by the salt;
    the resulting bit positions are set in the filter."""
    per_feature_size = _compute_per_feature_size(
        parameterization, feature_name)
    if per_feature_size == 0:
        return bytearray(0)

    feature_filter = _bit_array(per_feature_size)

    if not value:
        # Missing feature gets an empty filter; the matcher's
        # Fellegi-Sunter combiner handles missing-feature cases
        # under specific weights. Production has explicit
        # missing-feature sentinel handling; the demo uses an
        # empty filter as the simplest stand-in.
        return feature_filter

    # Tokenize into n-grams.
    n_grams = _produce_n_grams(
        value, n=parameterization["n_gram_size"],
        include_markers=True)

    # Hash each n-gram by k functions and set the bit positions.
    k = parameterization["hash_function_count"]
    for n_gram in n_grams:
        for k_index in range(k):
            position = _hash_to_bit_position(
                salt, k_index, n_gram, per_feature_size)
            _set_bit(feature_filter, position)

    return feature_filter

def _combine_per_feature_filters(per_feature_filters: dict,
                                      parameterization: dict) -> bytearray:
    """Combine the per-feature filters into the record-level
    CLK by concatenation in a fixed order. Production uses a more
    sophisticated combination scheme (interleaving with random
    hashing) that the simple concatenation in the demo
    illustrates the structure of."""
    total_size = parameterization["bloom_filter_size"]
    clk = _bit_array(total_size)
    cursor = 0
    # Iterate features in a deterministic order so all participants
    # produce comparable CLKs.
    for feature_name in sorted(parameterization[
                                    "per_feature_bit_allocation"].keys()):
        feature_size = _compute_per_feature_size(
            parameterization, feature_name)
        feature_filter = per_feature_filters.get(
            feature_name, _bit_array(feature_size))
        # Copy the feature filter's bits into the CLK at cursor.
        for i in range(feature_size):
            if cursor + i >= total_size:
                break
            if _is_bit_set(feature_filter, i):
                _set_bit(clk, cursor + i)
        cursor += feature_size
    return clk

def encode_record(prepared_record: dict,
                     parameterization_version: str,
                     salt_key_version: str,
                     cycle_id: str) -> dict:
    """
    Produce the encoded-record envelope. The envelope carries
    the CLK plus the metadata the matcher needs (participant_id,
    encoded_record_id, consent_posture, cohort_axis_hashes,
    parameterization_version) but does not carry the normalized
    demographics or the source-record identifier.
    """
    # 2A: load the pinned parameterization.
    parameterization = protocol_config_store.load_parameterization(
        parameterization_version)

    # 2B: load the salt key for this cycle. Production's
    # equivalent never returns plaintext; it returns an HSM
    # context that the encoder calls into for HMAC operations.
    salt = salt_custody.get_salt(
        salt_key_version, caller_role="encoder", cycle_id=cycle_id)

    # 2C: produce the per-feature Bloom filters.
    per_feature_filters = {}
    normalized = prepared_record["normalized_features"]
    for feature_name in parameterization[
                            "per_feature_bit_allocation"].keys():
        value = normalized.get(feature_name, "")
        per_feature_filters[feature_name] = (
            _encode_per_feature_bloom_filter(
                value, feature_name, parameterization, salt))

    # 2D: combine into the record-level CLK.
    clk = _combine_per_feature_filters(
        per_feature_filters, parameterization)

    # 2E: build the encoded-record envelope. Generate a per-cycle
    # pseudonym for the encoded_record_id so it cannot be
    # correlated across cycles by an observer; the participant
    # retains the source_record_id-to-encoded_record_id mapping
    # locally for its own lookups.
    encoded_record_id = (
        f"enc-{cycle_id}-"
        f"{prepared_record['participant_id'][:1]}-"
        f"{uuid.uuid4().hex[:12]}")

    encoded_record_envelope = {
        "participant_id":          prepared_record["participant_id"],
        "encoded_record_id":       encoded_record_id,
        "clk_payload":             bytes(clk),  # serialized as base64 in DDB/S3
        "per_feature_filters":     {  # retained for per-feature scoring
            f: bytes(filt) for f, filt in per_feature_filters.items()
        },
        "consent_posture":         prepared_record["consent_posture"],
        "cohort_axis_hashes":      prepared_record["cohort_axis_hashes"],
        "applicable_overlays":     prepared_record["applicable_overlays"],
        "parameterization_version": parameterization_version,
        "salt_key_version":        salt_key_version,
        "cycle_id":                cycle_id,
        "encoded_at":              _now_iso(),
    }

    # 2F: persist the per-cycle local mapping from the
    # encoded_record_id to the source_record_id. The mapping is
    # retained at the participant only; it is never included in
    # the encoded-record envelope or the cross-participant
    # exchange. The mapping enables the participant to later
    # resolve match results back to the source records under its
    # own access controls.
    _IN_MEMORY_PARTICIPANT_LOCAL_MAPPINGS.setdefault(
        prepared_record["participant_id"], {}
    )[encoded_record_id] = {
        "cycle_id":          cycle_id,
        "source_record_id":  prepared_record["source_record_id"],
        "encoded_at":        encoded_record_envelope["encoded_at"],
    }

    return encoded_record_envelope

def encode_batch(prepared_records: list,
                   parameterization_version: str,
                   salt_key_version: str,
                   cycle_id: str) -> list:
    """Bulk-encode a batch of prepared records. Production runs
    this in a Glue job with Spark partitioning; the demo iterates
    in-process."""
    encoded = [encode_record(r, parameterization_version,
                              salt_key_version, cycle_id)
                for r in prepared_records]
    _emit_metric("EncodedRecords", float(len(encoded)),
                  dimensions={"ParticipantId":
                                  prepared_records[0]["participant_id"]
                                  if prepared_records else "unknown",
                                "CycleId": cycle_id})
    return encoded

def exchange_encoded_records(encoded_record_envelopes: list,
                                  participant_id: str,
                                  cycle_id: str,
                                  trust_architecture: str = "linkage_broker_model"
                                  ) -> dict:
    """
    Validate that every envelope in the batch was produced under
    the parameterization and salt-key versions pinned for this
    cycle. Route to the linkage-execution endpoint per the trust
    architecture. Log the exchange to the participant's audit
    store.

    The demo runs all participants in one process so the
    "exchange" is logically rather than physically separate;
    production exchanges across AWS accounts and (often)
    organizational boundaries with cross-account bucket policies,
    PrivateLink, and mTLS.
    """
    # 3A: validate that every envelope's parameterization and
    # salt-key version match the cycle's pinned versions.
    # Mis-coordinated versions are the most common operational
    # failure mode for PPRL; catch them at exchange time before
    # the matcher silently returns zero matches.
    cycle_metadata = _IN_MEMORY_CYCLE_METADATA.get(cycle_id)
    if not cycle_metadata:
        raise RuntimeError(
            f"cycle metadata for {cycle_id} not found; "
            f"cycle must be initialized before exchange")

    expected_param_version = cycle_metadata["parameterization_version"]
    expected_salt_version  = cycle_metadata["salt_key_version"]

    for envelope in encoded_record_envelopes:
        if envelope["parameterization_version"] != expected_param_version:
            raise ValueError(
                f"parameterization mismatch for "
                f"{envelope['encoded_record_id']}: envelope "
                f"version {envelope['parameterization_version']} "
                f"does not match cycle's pinned version "
                f"{expected_param_version}. This is the silent-"
                f"failure mode the version-pinning is designed "
                f"to catch; do not proceed.")
        if envelope["salt_key_version"] != expected_salt_version:
            raise ValueError(
                f"salt-key mismatch for "
                f"{envelope['encoded_record_id']}: envelope "
                f"version {envelope['salt_key_version']} "
                f"does not match cycle's pinned version "
                f"{expected_salt_version}. The cycle's salt-key "
                f"version may have rotated mid-cycle; re-encode "
                f"under the active salt and re-run the cycle.")

    # 3B: route the envelopes to the linkage-execution endpoint
    # per the trust architecture. The demo's
    # _IN_MEMORY_EXCHANGE_BUCKET stands in for the cross-account
    # S3 buckets in the linkage-broker model, the tokenizer
    # ingestion endpoint in the tokenizer model, the Nitro
    # Enclave attested endpoint in the TEE model, or the SMPC
    # protocol runner in the SMPC model.
    bucket_for_participant = {
        "participant-A": PARTICIPANT_A_BUCKET,
        "participant-B": PARTICIPANT_B_BUCKET,
    }.get(participant_id, "unknown-bucket")

    transport_metadata = {
        "trust_architecture":  trust_architecture,
        "destination_bucket":  bucket_for_participant,
        "envelope_count":      len(encoded_record_envelopes),
    }

    # In production, upload to S3 with mTLS-and-cross-account
    # bucket policies. The demo just appends to the in-memory
    # exchange bucket.
    _IN_MEMORY_EXCHANGE_BUCKET[cycle_id].setdefault(
        participant_id, []).extend(encoded_record_envelopes)

    # 3C: log the exchange to the participant's audit store.
    audit_log_entry = {
        "event_type":              "PPRL_EXCHANGE",
        "cycle_id":                cycle_id,
        "participant_id":          participant_id,
        "parameterization_version": expected_param_version,
        "salt_key_version":        expected_salt_version,
        "record_count":            len(encoded_record_envelopes),
        "trust_architecture":      trust_architecture,
        "destination":             bucket_for_participant,
        "exchanged_at":            _now_iso(),
    }
    _archive_to_s3(audit_log_entry, AUDIT_ARCHIVE_BUCKET,
                     f"exchange-events/{cycle_id}",
                     key_id=f"{participant_id}-exchange")

    _emit_metric("ExchangeCompleted", 1.0,
                  dimensions={"ParticipantId":     participant_id,
                                "CycleId":           cycle_id,
                                "TrustArchitecture": trust_architecture})

    return transport_metadata

# In-memory exchange surface. Production replaces with cross-
# account S3 buckets behind PrivateLink endpoints.
_IN_MEMORY_EXCHANGE_BUCKET: dict = {}
_IN_MEMORY_CYCLE_METADATA:  dict = {}

def _candidate_pairs(encoded_records_a: list,
                       encoded_records_b: list) -> list:
    """Generate candidate pairs for matching. Production uses
    locality-sensitive hashing on the CLK to reduce O(n*m) to a
    tractable subset; the demo iterates the full Cartesian
    product because the demo population is small."""
    return [(a, b) for a in encoded_records_a
                    for b in encoded_records_b]

def _per_feature_bloom_filter_from_envelope(envelope: dict,
                                                  feature_name: str
                                                  ) -> bytearray:
    """Pull the per-feature filter from the encoded envelope.
    Production stores the per-feature filters separately so the
    matcher can do per-feature similarity scoring; the demo
    keeps them on the envelope."""
    raw = envelope["per_feature_filters"].get(feature_name)
    return bytearray(raw) if raw else bytearray(0)

def _compute_per_feature_similarities(envelope_a: dict,
                                          envelope_b: dict,
                                          parameterization: dict
                                          ) -> dict:
    """Per-feature Sørensen-Dice similarity. Each feature's
    Bloom filter is compared independently; the Fellegi-Sunter
    combiner then weights them."""
    similarities = {}
    for feature_name in parameterization[
                              "per_feature_bit_allocation"].keys():
        filter_a = _per_feature_bloom_filter_from_envelope(
            envelope_a, feature_name)
        filter_b = _per_feature_bloom_filter_from_envelope(
            envelope_b, feature_name)
        if len(filter_a) == 0 and len(filter_b) == 0:
            similarities[feature_name] = Decimal("0.5")  # missing both
        elif len(filter_a) == 0 or len(filter_b) == 0:
            similarities[feature_name] = Decimal("0.0")  # one missing
        else:
            similarities[feature_name] = _sorensen_dice_coefficient(
                filter_a, filter_b)
    return similarities

def _combine_with_fellegi_sunter(per_feature_similarities: dict,
                                      feature_weights: dict) -> Decimal:
    """Weighted sum across per-feature similarities. The
    production Fellegi-Sunter implementation uses log-likelihood
    ratios with EM-trained per-feature m-and-u parameters; the
    demo uses a simpler weighted-average that illustrates the
    structure without the parameter-estimation complexity."""
    total_weight = Decimal("0")
    weighted_sum = Decimal("0")
    for feature_name, similarity in per_feature_similarities.items():
        weight = feature_weights.get(feature_name, Decimal("0"))
        weighted_sum += weight * similarity
        total_weight += weight
    if total_weight == 0:
        return Decimal("0")
    return weighted_sum / total_weight

def match_encoded_records(encoded_records_a: list,
                              encoded_records_b: list,
                              cycle_id: str,
                              parameterization_version: str
                              ) -> list:
    """
    Compute pairwise similarity between encoded records under
    the protocol's matching function. Apply the encoded-data
    thresholds. Return the list of match decisions.
    """
    parameterization = protocol_config_store.load_parameterization(
        parameterization_version)

    # 4A: candidate generation.
    candidates = _candidate_pairs(encoded_records_a, encoded_records_b)

    match_results = []
    for record_a, record_b in candidates:
        # 4B: per-feature similarity.
        per_feature_similarities = _compute_per_feature_similarities(
            record_a, record_b, parameterization)

        # 4C: Fellegi-Sunter combination.
        match_score = _combine_with_fellegi_sunter(
            per_feature_similarities, FEATURE_WEIGHTS)

        # 4D: apply encoded-data thresholds.
        if match_score >= ENCODED_MATCH_HIGH_THRESHOLD:
            decision = "MATCH_HIGH"
        elif match_score >= ENCODED_MATCH_MED_THRESHOLD:
            decision = "MATCH_MED_REVIEW"
        elif match_score <= ENCODED_REJECT_THRESHOLD:
            decision = "REJECT"
        else:
            decision = "REVIEW"

        # 4E: build the per-pair result with evidence summary
        # and cohort-axis hashes for downstream cohort-stratified-
        # accuracy monitoring. Note: the result carries no
        # demographic features; just the encoded_record_ids and
        # the cohort-axis hashes.
        match_result = {
            "cycle_id":               cycle_id,
            "match_id":               f"match-{cycle_id}-"
                                       f"{uuid.uuid4().hex[:12]}",
            "participant_a_id":       record_a["participant_id"],
            "participant_b_id":       record_b["participant_id"],
            "encoded_record_a_id":    record_a["encoded_record_id"],
            "encoded_record_b_id":    record_b["encoded_record_id"],
            "match_score":            match_score,
            "decision":               decision,
            "evidence_summary": {
                "per_feature_similarities":  per_feature_similarities,
                "feature_weights_version":   MATCHER_CONFIG_VERSION,
            },
            "cohort_axis_hashes_a":   record_a["cohort_axis_hashes"],
            "cohort_axis_hashes_b":   record_b["cohort_axis_hashes"],
            "consent_posture_a":      record_a["consent_posture"],
            "consent_posture_b":      record_b["consent_posture"],
            "applicable_overlays_a":  record_a.get("applicable_overlays") or [],
            "applicable_overlays_b":  record_b.get("applicable_overlays") or [],
            "parameterization_version": parameterization_version,
            "matched_at":             _now_iso(),
        }
        match_results.append(match_result)

        # Cohort-stratified metric emit. The CohortBucketHash
        # dimension is what makes per-cohort linkage-rate
        # disparity monitoring possible without the matcher
        # learning the underlying axis values.
        cohort_hash = record_a["cohort_axis_hashes"].get(
            "name_tradition_hash", "unknown")
        _emit_metric("MatchDecision", 1.0,
                      dimensions={"Decision":         decision,
                                    "CohortBucketHash": cohort_hash[:8],
                                    "CycleId":          cycle_id})

    return match_results

def _filter_by_consent(match_results: list, purpose: str) -> list:
    """Drop matches whose consent posture does not permit
    inclusion in this disclosure. Records that the patient has
    not consented to include in this specific use case are
    excluded from the disclosure even if the matcher scored a
    match."""
    filtered = []
    for match_result in match_results:
        consent_a = match_result.get("consent_posture_a") or {}
        consent_b = match_result.get("consent_posture_b") or {}
        if purpose == "research_linkage":
            if (consent_a.get("consent_for_research_linkage")
                    and consent_b.get("consent_for_research_linkage")):
                filtered.append(match_result)
        elif purpose == "outcomes_research":
            if (consent_a.get("consent_for_outcomes_research_purpose")
                    and consent_b.get("consent_for_outcomes_research_purpose")):
                filtered.append(match_result)
        else:
            filtered.append(match_result)
    return filtered

def _build_per_record_match_flags(matches: list,
                                       target_consumer: str) -> dict:
    """Per-record disclosure form: each consumer receives the
    match flags for its own records. The consumer is a
    participating organization and can only see matches
    involving its own records; this filter is enforced here."""
    if target_consumer == "participant-A":
        a_side_matches = [
            {
                "encoded_record_a_id":     m["encoded_record_a_id"],
                "matched_with_participant_id": m["participant_b_id"],
                "match_decision":          m["decision"],
                "match_score":             m["match_score"],
            }
            for m in matches
            if m["decision"] in {"MATCH_HIGH", "MATCH_MED_REVIEW"}
                and m["participant_a_id"] == "participant-A"
        ]
        return {
            "matches":                       a_side_matches,
            "total_records_contributed_by_a":
                len(set(m["encoded_record_a_id"] for m in matches
                          if m["participant_a_id"] == "participant-A")),
            "total_matches_for_a":           len(a_side_matches),
        }
    if target_consumer == "participant-B":
        b_side_matches = [
            {
                "encoded_record_b_id":     m["encoded_record_b_id"],
                "matched_with_participant_id": m["participant_a_id"],
                "match_decision":          m["decision"],
                "match_score":             m["match_score"],
            }
            for m in matches
            if m["decision"] in {"MATCH_HIGH", "MATCH_MED_REVIEW"}
                and m["participant_b_id"] == "participant-B"
        ]
        return {
            "matches":                       b_side_matches,
            "total_records_contributed_by_b":
                len(set(m["encoded_record_b_id"] for m in matches
                          if m["participant_b_id"] == "participant-B")),
            "total_matches_for_b":           len(b_side_matches),
        }
    return {"matches": [], "note": "unknown target consumer"}

def _build_intersection_count(matches: list) -> dict:
    """Intersection-count disclosure form: only the size of the
    intersection (no per-record detail). Used when the protocol
    authorizes only the count, e.g., for public-health
    surveillance scenarios where the analytic question is
    'how many people are in both populations?' rather than
    'which specific people?'"""
    confirmed_matches = [m for m in matches
                          if m["decision"] == "MATCH_HIGH"]
    a_count = len(set(m["encoded_record_a_id"]
                        for m in confirmed_matches))
    b_count = len(set(m["encoded_record_b_id"]
                        for m in confirmed_matches))
    return {
        "intersection_count":  len(confirmed_matches),
        "participant_a_total": len(set(m["encoded_record_a_id"]
                                          for m in matches)),
        "participant_b_total": len(set(m["encoded_record_b_id"]
                                          for m in matches)),
        "match_rate_lower_bound":
            float(Decimal(a_count)
                    / Decimal(max(len(set(m["encoded_record_a_id"]
                                            for m in matches)), 1))),
        "match_rate_upper_bound":
            float(Decimal(b_count)
                    / Decimal(max(len(set(m["encoded_record_b_id"]
                                            for m in matches)), 1))),
    }

def _build_k_anonymous_aggregate(matches: list,
                                      k_parameter: int,
                                      suppression_threshold: int
                                      ) -> dict:
    """K-anonymous aggregate disclosure form: cohort-level
    aggregates with small-cell suppression. Used when the
    protocol authorizes only cohort summaries, e.g., for ACO
    out-of-network ED visit reporting where the consumer needs
    cohort-stratified visit counts but not per-patient detail."""
    confirmed_matches = [m for m in matches
                          if m["decision"] == "MATCH_HIGH"]

    # Aggregate by cohort axis hash (not the underlying values;
    # the matcher does not have visibility into them).
    by_cohort: dict = {}
    for match_result in confirmed_matches:
        bucket_hash = match_result["cohort_axis_hashes_a"].get(
            "name_tradition_hash", "unknown")[:8]
        by_cohort.setdefault(bucket_hash, 0)
        by_cohort[bucket_hash] += 1

    aggregates = []
    for bucket_hash, count in sorted(by_cohort.items()):
        if count < suppression_threshold:
            aggregates.append({
                "cohort_axis_hash_prefix": bucket_hash,
                "patient_count_aggregate": "[suppressed: cell size below threshold]",
            })
        else:
            aggregates.append({
                "cohort_axis_hash_prefix": bucket_hash,
                "patient_count_aggregate": count,
            })
    return {
        "k_anonymity_parameter":  k_parameter,
        "suppression_threshold":  suppression_threshold,
        "cohort_aggregates":      aggregates,
    }

def disclose_linkage_results(match_results: list,
                                  cycle_id: str,
                                  disclosure_form: str,
                                  target_consumer: str,
                                  purpose: str = "research_linkage",
                                  k_parameter: int = DEFAULT_K_ANONYMITY,
                                  suppression_threshold: int = DEFAULT_SUPPRESSION_THRESHOLD
                                  ) -> dict:
    """
    Apply the disclosure-policy form and route the linkage
    results to the target consumer. Emit the cycle-completion
    event for cross-recipe consumers.
    """
    # 5A: filter by consent posture.
    consent_filtered = _filter_by_consent(match_results, purpose)
    consent_filtered_count = len(consent_filtered)
    consent_dropped = len(match_results) - consent_filtered_count

    # 5B: apply the disclosure-form transformation.
    if disclosure_form == "per_record_match_flags":
        envelope_payload = _build_per_record_match_flags(
            consent_filtered, target_consumer)
    elif disclosure_form == "intersection_count":
        envelope_payload = _build_intersection_count(consent_filtered)
    elif disclosure_form == "k_anonymous_aggregate":
        envelope_payload = _build_k_anonymous_aggregate(
            consent_filtered, k_parameter, suppression_threshold)
    elif disclosure_form not in DISCLOSURE_FORMS:
        raise ValueError(
            f"unknown disclosure_form {disclosure_form}; "
            f"authorized forms: {sorted(DISCLOSURE_FORMS)}")
    else:
        envelope_payload = {
            "note": f"disclosure_form {disclosure_form} not "
                     f"implemented in demo; production uses real "
                     f"DP / encryption primitives"
        }

    disclosure_envelope = {
        "cycle_id":          cycle_id,
        "disclosure_form":   disclosure_form,
        "target_consumer":   target_consumer,
        "purpose":           purpose,
        "envelope_payload":  envelope_payload,
        "consent_dropped":   consent_dropped,
        "disclosed_at":      _now_iso(),
    }

    # 5C: route to the target consumer. Production uses signed
    # envelopes with mTLS; the demo logs the routing.
    logger.info("disclosure routed",
                  extra={"target_consumer": target_consumer,
                          "disclosure_form": disclosure_form})

    # 5D: emit the cycle-completion event for cross-recipe
    # consumers.
    try:
        eventbridge_client.put_events(Entries=[{
            "Source":       "privacy-preserving-record-linkage",
            "DetailType":   "pprl_linkage_cycle_completed",
            "EventBusName": PPRL_EVENT_BUS_NAME,
            "Detail": json.dumps({
                "cycle_id":          cycle_id,
                "disclosure_form":   disclosure_form,
                "target_consumer":   target_consumer,
                "matched_pair_count":
                    len([m for m in consent_filtered
                          if m["decision"] == "MATCH_HIGH"]),
                "review_pair_count":
                    len([m for m in consent_filtered
                          if m["decision"] == "MATCH_MED_REVIEW"]),
                "disclosed_at":      _now_iso(),
            }, default=str),
        }])
    except Exception as exc:
        logger.info("event emit skipped (demo mode is fine to ignore)",
                     extra={"error": str(exc)})

    # 5E: log the disclosure to the audit archive. The audit log
    # captures the disclosure-form, the target-consumer, the
    # per-record count, and the cohort-stratified-accuracy
    # summary, but does NOT log the actual disclosure payload
    # (which would duplicate the consumer's copy).
    audit_entry = {
        "event_type":        "PPRL_DISCLOSURE",
        "cycle_id":          cycle_id,
        "disclosure_form":   disclosure_form,
        "target_consumer":   target_consumer,
        "purpose":           purpose,
        "record_count":      consent_filtered_count,
        "consent_dropped":   consent_dropped,
        "disclosed_at":      _now_iso(),
    }
    _archive_to_s3(audit_entry, AUDIT_ARCHIVE_BUCKET,
                     f"disclosure-events/{cycle_id}",
                     key_id=f"{target_consumer}-disclosure")

    _emit_metric("DisclosureCompleted", 1.0,
                  dimensions={"DisclosureForm":  disclosure_form,
                                "TargetConsumer":  target_consumer,
                                "CycleId":         cycle_id})
    return disclosure_envelope

def invalidate_on_event(invalidation_event: dict) -> dict:
    """Re-evaluate linkages affected by an invalidation event.
    Production triggers a Step Functions workflow that handles
    the specific superseding-event type and re-runs the affected
    portions of the cycle. The demo records what would be done."""
    source = invalidation_event.get("source")
    summary = {
        "source":              source,
        "event_id":            invalidation_event.get("event_id"),
        "actions":             [],
        "affected_cycle_ids":  [],
        "affected_participants": [],
    }

    if source == "salt_rotation":
        # The salt has rotated. All encoded data under the prior
        # salt is invalidated. Schedule a coordinated re-encoding
        # cycle with all participants. Notify downstream consumers
        # that the prior cycle's results are superseded by the
        # next cycle's.
        prior_salt = invalidation_event.get("prior_salt_key_version")
        new_salt = invalidation_event.get("new_salt_key_version")
        approvers = invalidation_event.get("dual_control_approvers", [])

        try:
            salt_custody.rotate_salt(new_salt, approvers)
        except ValueError as exc:
            summary["actions"].append(f"rotation_blocked: {exc}")
            return summary

        affected = [cid for cid, meta in _IN_MEMORY_CYCLE_METADATA.items()
                     if meta.get("salt_key_version") == prior_salt]
        summary["affected_cycle_ids"] = affected
        summary["actions"].append(
            f"rotate_salt_to:{new_salt}")
        summary["actions"].append(
            f"schedule_coordinated_re_encoding:{len(affected)}_cycles")
        summary["actions"].append(
            "notify_downstream_consumers_of_supersession")

    elif source == "parameterization_upgrade":
        # The parameterization has been upgraded. Schedule a
        # coordinated re-encoding under the new parameterization.
        # Existing cycles' results remain valid for the use cases
        # the prior parameterization supported, but new cycles
        # use the new parameterization.
        new_version = invalidation_event.get("new_parameterization_version")
        new_config = invalidation_event.get("new_parameterization")
        if new_version and new_config:
            protocol_config_store.publish_new_version(
                new_version, new_config)
        summary["actions"].append(
            f"publish_new_parameterization:{new_version}")
        summary["actions"].append(
            "schedule_coordinated_re_encoding")

    elif source == "consent_withdrawal":
        # A patient has withdrawn consent for inclusion in the
        # linkage. The participating organization removes the
        # patient from future encoding. Prior linkages remain in
        # the consumer's possession; the participating
        # organization communicates the withdrawal to the
        # consumer per the protocol's policy on retroactive
        # handling.
        sid = invalidation_event.get("source_record_id")
        participant = invalidation_event.get("participant_id")
        consent_store.withdraw_consent(sid)
        summary["affected_participants"] = [participant]
        summary["actions"].append(
            f"mark_consent_withdrawn:{sid}")
        summary["actions"].append(
            "notify_downstream_consumers_of_withdrawal")
        summary["actions"].append(
            "exclude_from_future_cycles")

    elif source == "identity_merge_recipe_5_1":
        # Recipe 5.1 merged two identities at the participating
        # organization. The prior encoded records under the
        # merged-from identity are invalidated; the next cycle
        # re-encodes the surviving identity.
        participant = invalidation_event.get("participant_id")
        merged_from_ids = invalidation_event.get(
            "merged_from_source_record_ids", [])
        summary["affected_participants"] = [participant]
        summary["actions"].append(
            f"queue_re_encode_for_merged_identities:"
            f"{len(merged_from_ids)}_records")

    elif source == "name_change_recipe_5_7":
        # Recipe 5.7 resolved a name change. The encoded records
        # under the prior name are invalidated; the next cycle
        # re-encodes under the current name (and may carry a
        # prior-name encoding for the cycles that need
        # historical-name matching).
        participant = invalidation_event.get("participant_id")
        sid = invalidation_event.get("source_record_id")
        summary["affected_participants"] = [participant]
        summary["actions"].append(
            f"queue_re_encode_for_name_change:{sid}")

    elif source == "re_identification_risk_model_update":
        # The institutional privacy team has updated the
        # re-identification-risk model. The parameterization may
        # need re-tuning; defensive measures may need to be
        # strengthened. Schedule a coordinated parameterization
        # upgrade and re-encoding cycle.
        summary["actions"].append(
            "schedule_parameterization_upgrade")
        summary["actions"].append(
            "schedule_coordinated_re_encoding")

    else:
        summary["actions"].append(
            f"unknown_invalidation_source:{source}")

    # Emit the invalidation event for downstream consumers.
    try:
        eventbridge_client.put_events(Entries=[{
            "Source":       "privacy-preserving-record-linkage",
            "DetailType":   "pprl_linkage_invalidated",
            "EventBusName": PPRL_EVENT_BUS_NAME,
            "Detail": json.dumps({
                "invalidation_source":       source,
                "invalidation_event_id":     invalidation_event.get("event_id"),
                "affected_cycle_ids":        summary["affected_cycle_ids"],
                "affected_participant_ids":  summary["affected_participants"],
                "invalidated_at":            _now_iso(),
            }, default=str),
        }])
    except Exception as exc:
        logger.info("invalidation event emit skipped (demo mode)",
                     extra={"error": str(exc)})

    _emit_metric("Invalidations", 1.0,
                  dimensions={"Source": source or "unknown"})
    return summary

def initialize_cycle(cycle_id: str,
                       parameterization_version: str = None,
                       salt_key_version: str = None) -> dict:
    """Set up the linkage cycle metadata. Each cycle pins the
    parameterization version and the salt-key version active at
    the cycle's start; mismatched versions at exchange time
    trigger the silent-failure-prevention check from Step 3."""
    parameterization_version = (parameterization_version
                                  or PROTOCOL_PARAMETERIZATION_VERSION)
    salt_key_version = (salt_key_version
                         or salt_custody.get_active_version())
    cycle_metadata = {
        "cycle_id":                cycle_id,
        "parameterization_version": parameterization_version,
        "salt_key_version":        salt_key_version,
        "started_at":              _now_iso(),
        "participants":            ["participant-A", "participant-B"],
    }
    _IN_MEMORY_CYCLE_METADATA[cycle_id] = cycle_metadata
    _IN_MEMORY_EXCHANGE_BUCKET.setdefault(cycle_id, {})
    return cycle_metadata

def run_cycle(cycle_id: str,
                participant_a_records: list,
                participant_b_records: list,
                disclosure_form: str = "per_record_match_flags",
                target_consumer: str = "participant-A",
                purpose: str = "research_linkage") -> dict:
    """End-to-end standardize -> encode -> exchange -> match ->
    disclose pipeline for a single linkage cycle. Returns the
    full trace."""
    cycle_metadata = initialize_cycle(cycle_id)
    parameterization_version = cycle_metadata["parameterization_version"]
    salt_key_version = cycle_metadata["salt_key_version"]

    # Per-participant standardize-and-encode-and-exchange. In
    # production each participant runs this independently in its
    # own AWS account; the demo runs both sequentially in one
    # process.
    prepared_a = standardize_and_prepare(
        participant_a_records, "participant-A",
        purpose, PROTOCOL_PARAMETERIZATION)
    encoded_a = encode_batch(prepared_a, parameterization_version,
                              salt_key_version, cycle_id)
    exchange_meta_a = exchange_encoded_records(
        encoded_a, "participant-A", cycle_id)

    prepared_b = standardize_and_prepare(
        participant_b_records, "participant-B",
        purpose, PROTOCOL_PARAMETERIZATION)
    encoded_b = encode_batch(prepared_b, parameterization_version,
                              salt_key_version, cycle_id)
    exchange_meta_b = exchange_encoded_records(
        encoded_b, "participant-B", cycle_id)

    # Match. In production this runs in the linkage-execution
    # endpoint's account (a Nitro Enclave for TEE-based protocols
    # or a Glue job for the linkage-broker model).
    match_results = match_encoded_records(
        encoded_a, encoded_b, cycle_id, parameterization_version)
    _IN_MEMORY_LINKAGE_RESULTS.extend(match_results)

    # Disclose. The disclosure form is specified per use case;
    # the demo varies it across cycles.
    disclosure_envelope = disclose_linkage_results(
        match_results, cycle_id, disclosure_form,
        target_consumer, purpose)

    return {
        "cycle_metadata":     cycle_metadata,
        "encoded_a_count":    len(encoded_a),
        "encoded_b_count":    len(encoded_b),
        "match_results":      match_results,
        "disclosure_envelope": disclosure_envelope,
    }

def run_demo():
    """Run three representative linkage cycles covering the
    per-record-match-flags, intersection-count, and k-anonymous-
    aggregate disclosure forms. Then exercise the invalidation
    pipeline."""
    print("=" * 72)
    print("Privacy-Preserving Record Linkage Demo")
    print("=" * 72)
    print()
    print("All patients, demographics, and consent records in this demo are")
    print("fictional. The mock salt custody, protocol parameterization store,")
    print("participant data sources, consent store, and jurisdictional")
    print("overlays return hand-crafted data that exercises the encoding,")
    print("matching, and disclosure paths; do not point this demo at a live")
    print("data-sharing collaboration.")
    print()
    print(f"Active parameterization: {PROTOCOL_PARAMETERIZATION_VERSION}")
    print(f"Active salt key:         {salt_custody.get_active_version()}")
    print(f"Bloom filter size:       "
          f"{PROTOCOL_PARAMETERIZATION['bloom_filter_size']} bits")
    print(f"Hash function count:     "
          f"{PROTOCOL_PARAMETERIZATION['hash_function_count']}")
    print(f"N-gram size:             "
          f"{PROTOCOL_PARAMETERIZATION['n_gram_size']}")
    print(f"Encoded thresholds:      "
          f"HIGH={ENCODED_MATCH_HIGH_THRESHOLD}  "
          f"MED={ENCODED_MATCH_MED_THRESHOLD}  "
          f"REJECT={ENCODED_REJECT_THRESHOLD}")
    print()

    # --- Cycle 1: per_record_match_flags to participant A ---
    print("-" * 72)
    print("Cycle 1: research linkage, per_record_match_flags to participant A")
    print("-" * 72)
    cycle_1 = run_cycle(
        cycle_id="cycle-2026-q2-research-001",
        participant_a_records=SYNTHETIC_PARTICIPANT_A_RECORDS,
        participant_b_records=SYNTHETIC_PARTICIPANT_B_RECORDS,
        disclosure_form="per_record_match_flags",
        target_consumer="participant-A",
        purpose="research_linkage",
    )
    print(f"  participants:         participant-A ({cycle_1['encoded_a_count']} encoded), "
          f"participant-B ({cycle_1['encoded_b_count']} encoded)")
    confirmed = [m for m in cycle_1["match_results"]
                  if m["decision"] == "MATCH_HIGH"]
    review = [m for m in cycle_1["match_results"]
                  if m["decision"] == "MATCH_MED_REVIEW"]
    print(f"  match_high pairs:     {len(confirmed)}")
    print(f"  review pairs:         {len(review)}")
    print(f"  disclosure_form:      {cycle_1['disclosure_envelope']['disclosure_form']}")
    print(f"  target_consumer:      {cycle_1['disclosure_envelope']['target_consumer']}")
    print(f"  consent_dropped:      {cycle_1['disclosure_envelope']['consent_dropped']}")
    payload = cycle_1["disclosure_envelope"]["envelope_payload"]
    print(f"  total_records_for_a:  {payload.get('total_records_contributed_by_a')}")
    print(f"  total_matches_for_a:  {payload.get('total_matches_for_a')}")

    # Show a sample match result with evidence summary so the
    # reader can see what the per-feature similarity scores look
    # like under encoded comparison.
    if confirmed:
        sample = confirmed[0]
        print(f"  sample MATCH_HIGH:")
        print(f"    encoded_record_a_id: {sample['encoded_record_a_id']}")
        print(f"    encoded_record_b_id: {sample['encoded_record_b_id']}")
        print(f"    match_score:         {float(sample['match_score']):.3f}")
        print(f"    per_feature_similarities (excerpt):")
        for fname in ["given_name", "family_name", "dob", "address_line"]:
            sim = sample["evidence_summary"][
                "per_feature_similarities"].get(fname)
            if sim is not None:
                print(f"      {fname:<14}: {float(sim):.3f}")

    # --- Cycle 2: intersection_count to a public-health consumer ---
    print()
    print("-" * 72)
    print("Cycle 2: public-health surveillance, intersection_count")
    print("-" * 72)
    cycle_2 = run_cycle(
        cycle_id="cycle-2026-q2-surveillance-002",
        participant_a_records=SYNTHETIC_PARTICIPANT_A_RECORDS,
        participant_b_records=SYNTHETIC_PARTICIPANT_B_RECORDS,
        disclosure_form="intersection_count",
        target_consumer="state-public-health-department",
        purpose="research_linkage",
    )
    payload_2 = cycle_2["disclosure_envelope"]["envelope_payload"]
    print(f"  intersection_count:        {payload_2.get('intersection_count')}")
    print(f"  participant_a_total:       {payload_2.get('participant_a_total')}")
    print(f"  participant_b_total:       {payload_2.get('participant_b_total')}")
    print(f"  match_rate_lower_bound:    "
          f"{payload_2.get('match_rate_lower_bound'):.3f}")
    print(f"  match_rate_upper_bound:    "
          f"{payload_2.get('match_rate_upper_bound'):.3f}")
    print(f"  consent_dropped:           "
          f"{cycle_2['disclosure_envelope']['consent_dropped']}")

    # --- Cycle 3: k_anonymous_aggregate ---
    print()
    print("-" * 72)
    print("Cycle 3: ACO out-of-network analytics, k_anonymous_aggregate")
    print("-" * 72)
    cycle_3 = run_cycle(
        cycle_id="cycle-2026-q2-aco-003",
        participant_a_records=SYNTHETIC_PARTICIPANT_A_RECORDS,
        participant_b_records=SYNTHETIC_PARTICIPANT_B_RECORDS,
        disclosure_form="k_anonymous_aggregate",
        target_consumer="aco-analytics-team",
        purpose="research_linkage",
    )
    payload_3 = cycle_3["disclosure_envelope"]["envelope_payload"]
    print(f"  k_anonymity_parameter:     {payload_3.get('k_anonymity_parameter')}")
    print(f"  suppression_threshold:     {payload_3.get('suppression_threshold')}")
    print(f"  cohort_aggregates:")
    for agg in payload_3.get("cohort_aggregates", []):
        print(f"    cohort_axis_hash_prefix={agg['cohort_axis_hash_prefix']}: "
              f"{agg['patient_count_aggregate']}")

    # --- Phase 2: invalidation pipeline ---
    print()
    print("-" * 72)
    print("Phase 2: invalidation triggers")
    print("-" * 72)

    # Salt rotation. Requires dual-control approval.
    inv_1 = invalidate_on_event({
        "source":                  "salt_rotation",
        "event_id":                "inv-2026-04-30-001",
        "prior_salt_key_version":  "salt-2026-q2-rotation-001",
        "new_salt_key_version":    "salt-2026-q3-rotation-001",
        "dual_control_approvers": [
            "alice@participant-A.org",
            "bob@participant-B.org",
        ],
    })
    print(f"  source={inv_1['source']:<35}"
          f" affected={len(inv_1['affected_cycle_ids'])} cycles")
    for action in inv_1['actions']:
        print(f"    action: {action}")

    # Salt rotation without dual control should fail.
    inv_2 = invalidate_on_event({
        "source":                  "salt_rotation",
        "event_id":                "inv-2026-04-30-002",
        "prior_salt_key_version":  "salt-2026-q3-rotation-001",
        "new_salt_key_version":    "salt-2026-q4-rotation-001",
        "dual_control_approvers": ["alice@participant-A.org"],  # only one
    })
    print(f"  source={inv_2['source']:<35} "
          f"actions={inv_2['actions']}")

    # Parameterization upgrade.
    new_param_config = dict(PROTOCOL_PARAMETERIZATION)
    new_param_config["parameterization_version"] = "pprl-clk-v2.4.0"
    new_param_config["defensive_measures"]["random_hashing_enabled"] = True
    inv_3 = invalidate_on_event({
        "source":                       "parameterization_upgrade",
        "event_id":                     "inv-2026-04-30-003",
        "new_parameterization_version": "pprl-clk-v2.4.0",
        "new_parameterization":         new_param_config,
    })
    print(f"  source={inv_3['source']:<35} "
          f"actions={inv_3['actions']}")

    # Consent withdrawal.
    inv_4 = invalidate_on_event({
        "source":           "consent_withdrawal",
        "event_id":         "inv-2026-04-30-004",
        "participant_id":   "participant-A",
        "source_record_id": "academic-mc-mrn-00284271",
    })
    print(f"  source={inv_4['source']:<35} "
          f"actions={inv_4['actions']}")

    # Identity merge from recipe 5.1.
    inv_5 = invalidate_on_event({
        "source":         "identity_merge_recipe_5_1",
        "event_id":       "inv-2026-04-30-005",
        "participant_id": "participant-A",
        "merged_from_source_record_ids": [
            "academic-mc-mrn-99999-old",
            "academic-mc-mrn-88888-old",
        ],
    })
    print(f"  source={inv_5['source']:<35} "
          f"actions={inv_5['actions']}")

    # Name change from recipe 5.7.
    inv_6 = invalidate_on_event({
        "source":           "name_change_recipe_5_7",
        "event_id":         "inv-2026-04-30-006",
        "participant_id":   "participant-A",
        "source_record_id": "academic-mc-mrn-00284271",
    })
    print(f"  source={inv_6['source']:<35} "
          f"actions={inv_6['actions']}")

if __name__ == "__main__":
    run_demo()