Recipe 3.9: Python Implementation Example

pip install boto3 scikit-learn pandas numpy joblib networkx

import io
import json
import logging
import math
import uuid
from collections import defaultdict, Counter
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Optional

import boto3
import joblib
import networkx as nx
import numpy as np
import pandas as pd
from botocore.config import Config
from boto3.dynamodb.conditions import Key
from sklearn.linear_model import LogisticRegression
from sklearn.isotonic import IsotonicRegression

# Structured logging. Ship JSON records to CloudWatch Logs Insights. Audit
# events, workforce identities, patient identities, break-glass reasons,
# and case narratives all contain PHI or workforce PII. Log structural
# metadata only. Never log full audit payloads with workforce-and-patient
# identifiers, full feature vectors, raw break-glass reason text, or
# Bedrock prompts in application logs. The audit indexes (OpenSearch) and
# the case-state store (DynamoDB) are the right home for full payloads,
# behind KMS and CloudTrail data events.
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Adaptive retry handles throttling across DynamoDB, Timestream, Kinesis,
# SageMaker, and Bedrock with exponential backoff and jitter. Audit-event
# ingest is bursty (EHRs flush large batches at quarter-hour or top-of-hour
# boundaries; IdP and VPN logs spike at shift change), and adaptive mode
# keeps burst windows from cascading into retry storms against the
# enrichment cache and the scoring endpoint.
BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 5, "mode": "adaptive"})

REGION = "us-east-1"
dynamodb = boto3.resource("dynamodb", region_name=REGION, config=BOTO3_RETRY_CONFIG)
kinesis = boto3.client("kinesis", region_name=REGION, config=BOTO3_RETRY_CONFIG)
timestream_write = boto3.client("timestream-write", region_name=REGION, config=BOTO3_RETRY_CONFIG)
timestream_query = boto3.client("timestream-query", region_name=REGION, config=BOTO3_RETRY_CONFIG)
s3_client = boto3.client("s3", region_name=REGION, config=BOTO3_RETRY_CONFIG)
cloudwatch = boto3.client("cloudwatch", region_name=REGION, config=BOTO3_RETRY_CONFIG)
eventbridge = boto3.client("events", region_name=REGION, config=BOTO3_RETRY_CONFIG)
featurestore_runtime = boto3.client(
    "sagemaker-featurestore-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG
)
sagemaker_runtime = boto3.client("sagemaker-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG)
bedrock_runtime = boto3.client("bedrock-runtime", region_name=REGION, config=BOTO3_RETRY_CONFIG)

# --- Resource Names ---
# Fill in with your actual resource names. These are placeholders.
WORKFORCE_IDENTITY_TABLE = "workforce-identity"
WORKFORCE_SCHEDULE_TABLE = "workforce-schedule"
PATIENT_CONTEXT_TABLE    = "patient-context"
USER_STATE_TABLE         = "user-state"
CASE_STATE_TABLE         = "case-state"
SUPPRESSION_RULES_TABLE  = "suppression-rules"

AUDIT_EVENTS_STREAM = "audit-events"
TIMESTREAM_DATABASE = "access-monitoring"
TIMESTREAM_BASELINE_TABLE = "user_behavioral_baselines"

ACCESS_ANOMALY_FEATURES_FG = "access-anomaly-features-online"

RAW_EVENTS_BUCKET     = "my-access-monitoring-raw-events"
TRAINING_LABELS_BUCKET = "my-access-monitoring-training-labels"
MODEL_ARTIFACTS_BUCKET = "my-access-monitoring-model-artifacts"

CASE_BUS                  = "access-anomaly-events"
SAGEMAKER_ENDPOINT_NAME    = "access-anomaly-composite-prod"
BEDROCK_MODEL_ID           = "anthropic.claude-3-sonnet-20240229-v1:0"

# --- Detector Weights ---
# The composite scorer is a weighted sum of per-detector outputs. Per-cohort
# tuning catches systematic differences in how each detector class performs
# for different role categories. Production loads these from a versioned
# table that the joint privacy-and-infosec governance committee owns.
DEFAULT_DETECTOR_WEIGHTS = {
    "DEFAULT": {"rules": 0.40, "baseline": 0.20, "graph": 0.30, "sequence": 0.10},
    # Privileged users (DBAs, integration engineers) get more weight on
    # graph and baseline, less on rules, because the rule library is
    # designed primarily for clinical workforce behavior.
    "privileged_user": {"rules": 0.20, "baseline": 0.30, "graph": 0.40, "sequence": 0.10},
    # Clinical workforce gets the standard mix. Same-name and VIP rules
    # are high-yield here.
    "clinical_workforce": {"rules": 0.45, "baseline": 0.20, "graph": 0.25, "sequence": 0.10},
    # New users (under 90 days) get reduced rules weight because their
    # search-and-explore behavior trips the baseline detector for
    # learning reasons, not snooping reasons.
    "new_user": {"rules": 0.50, "baseline": 0.10, "graph": 0.30, "sequence": 0.10},
}

# --- Tier Thresholds ---
# Tier thresholds are cohort-stratified in production. Different role
# categories have different base rates of confirmed violation and
# different review costs per case. Production loads thresholds from a
# versioned DynamoDB table so the governance committee can update them
# without a code deploy.
DEFAULT_TIER_THRESHOLDS = {
    "DEFAULT":            {"tier_1": 0.80, "tier_2": 0.60, "tier_3": 0.40},
    "clinical_workforce": {"tier_1": 0.78, "tier_2": 0.58, "tier_3": 0.38},
    "privileged_user":    {"tier_1": 0.85, "tier_2": 0.65, "tier_3": 0.45},
    "new_user":           {"tier_1": 0.82, "tier_2": 0.62, "tier_3": 0.42},
}

# --- Rule Severity Weights ---
# Each rule emits a severity. The composite layer maps severity to a
# numeric confidence floor that the rules-block contributes to the
# composite. High-severity rules with high confidence dominate; low-
# severity rules add modestly.
RULE_SEVERITY_FLOOR = {
    "high":              0.90,
    "medium":            0.70,
    "low":               0.45,
    "policy_dependent":  0.50,   # interpretation depends on org policy
}

# --- Window Sizes ---
# Trajectory and baseline windows. Matching the main recipe; tunable per
# cohort in production.
BASELINE_LOOKBACK_DAYS         = 30   # rolling window for the per-user behavioral baseline
PEER_GROUP_LOOKBACK_DAYS       = 60   # peer-group baseline window (longer; smooths noise)
MIN_BASELINE_DAYS              = 14   # below this, fall back to peer-group baseline only
ACTIVITY_AGGREGATION_WINDOWS   = [1, 8, 24, 168, 720]   # hours; 1h, 8h, 1d, 7d, 30d

# --- Suppression Windows ---
SUPPRESSION_AFTER_DISMISSAL_DAYS = 30
CASE_GROUPING_WINDOW_DAYS         = 7   # group same user-patient cases within this window

# --- Name Uniqueness Reference (toy, for the demo) ---
# Production uses a real surname-frequency reference (US Census or
# equivalent regional dataset). Higher uniqueness = stronger same-name
# signal. Smith and Johnson are very common; Wojnarowski is uncommon.
SURNAME_UNIQUENESS = {
    "smith":         0.05,
    "johnson":       0.06,
    "williams":      0.07,
    "garcia":        0.08,
    "miller":        0.09,
    "davis":         0.10,
    "wojnarowski":   0.92,
    "kovalenko":     0.88,
    "abernathy":     0.78,
    "okonkwo":       0.85,
    "fitzgerald":    0.55,
    "DEFAULT":       0.40,   # fallback when surname is not in the reference
}

# --- ZIP Population Density Reference (toy, for the demo) ---
# Production uses Census tract or ZCTA population data. Small ZIPs
# (rural, small-town) are higher-signal for same-neighborhood detection
# because the prior probability of two unrelated people sharing the ZIP
# is lower.
SMALL_ZIP_THRESHOLD = 15000   # population
ZIP_POPULATION = {
    "14620": 9400,    # small-zip example used in the recipe sample case
    "10001": 24000,
    "60601": 12500,
    "94110": 32000,
    "DEFAULT": 25000,
}

# --- Cohort Inference ---
# The user role and tenure determine which threshold and detector-weight
# cohort the user lands in. Production loads this mapping from a
# governance-approved table.
PRIVILEGED_ROLES = {
    "database_administrator", "integration_engineer", "data_engineer",
    "it_analyst", "system_administrator", "ehr_analyst",
}
CLINICAL_ROLES = {
    "registered_nurse", "physician", "resident", "fellow", "physician_assistant",
    "nurse_practitioner", "pharmacist", "respiratory_therapist", "medical_assistant",
    "social_worker", "case_manager",
}

def _to_decimal(value, precision="0.0001"):
    """Convert numeric input to Decimal for DynamoDB storage.

    DynamoDB rejects Python float for numeric attributes because float
    arithmetic introduces rounding drift that makes threshold comparisons
    unreliable over time. Always pass calibrated probabilities, deviation
    z-scores, and confidence values through Decimal on the way in and
    back out.
    """
    if value is None:
        return None
    return Decimal(str(value)).quantize(Decimal(precision))

def _decimalize(obj):
    """Recursively convert floats to Decimals for DynamoDB write."""
    if isinstance(obj, float):
        return _to_decimal(obj)
    if isinstance(obj, dict):
        return {k: _decimalize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_decimalize(v) for v in obj]
    return obj

def _undecimalize(obj):
    """Inverse of _decimalize for read-side conversion to Python-native types."""
    if isinstance(obj, Decimal):
        return float(obj)
    if isinstance(obj, dict):
        return {k: _undecimalize(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_undecimalize(v) for v in obj]
    return obj

def resolve_workforce_id(ehr_user_id, source_format):
    """Map EHR-internal user ID to enterprise identity.

    Production maintains a per-source mapping table from EHR user
    identifier to Active Directory SID, Okta user ID, or the equivalent
    enterprise identifier. The teaching example assumes the EHR user
    ID already matches the enterprise identifier.
    """
    # In production:
    #   table = dynamodb.Table("user-id-mapping")
    #   response = table.get_item(Key={"source_user_id": ehr_user_id,
    #                                  "source_system":  source_format})
    #   return response.get("Item", {}).get("workforce_id")
    return ehr_user_id

def resolve_patient_id(ehr_patient_id, source_format):
    """Map EHR-internal patient ID to enterprise master patient identifier.

    Production calls the EMPI service (Verato, NextGate, vendor-specific
    Mirth-driven matching) for cross-system patient identity. The
    teaching example assumes the EHR patient ID is the EMPI.
    """
    return ehr_patient_id

def normalize_event_type(action_type):
    """Normalize source-specific action verbs to a canonical set."""
    canonical = {
        "view":          "view",
        "open":          "view",
        "read":          "view",
        "edit":          "edit",
        "modify":        "edit",
        "update":        "edit",
        "print":         "print",
        "export":        "export",
        "download":      "export",
        "search":        "search",
        "login":         "login",
        "auth":          "login",
        "break_glass":   "break_glass",
    }
    return canonical.get(action_type.lower(), action_type.lower())

def normalize_resource_type(resource):
    """Normalize source-specific resource labels to a canonical set."""
    canonical = {
        "chart":          "chart",
        "encounter":      "chart",
        "demographic":    "demographics",
        "demographics":   "demographics",
        "medication":     "medication_list",
        "medications":    "medication_list",
        "lab":            "lab_result",
        "lab_result":     "lab_result",
        "image":          "image",
        "imaging":        "image",
        "note":           "note",
        "progress_note":  "note",
        "order":          "order",
        "discharge":      "discharge_summary",
        "discharge_summary": "discharge_summary",
    }
    return canonical.get(resource.lower(), resource.lower())

def on_ehr_audit_event(raw_event, source_format):
    """Receive an EHR audit event, normalize, and put on the event stream.

    Production wires this as a Lambda triggered by the EHR audit-feed
    connector. The Lambda parses the source-specific payload, performs
    identity and patient resolution, builds the canonical event, and
    writes to Kinesis. The downstream enrichment Lambda picks it up.
    """
    workforce_id = resolve_workforce_id(raw_event["user_id"], source_format)
    if workforce_id is None:
        # Unknown workforce user. Could be a service account that lost its
        # mapping, a stale identifier, or a misconfigured feed. Production
        # routes to a quarantine queue for investigation; the teaching
        # example logs and drops.
        logger.warning("unknown workforce user",
                       extra={"ehr_user_id": raw_event["user_id"],
                              "source_format": source_format})
        return {"statusCode": 202, "reason": "unknown_workforce_user"}

    patient_id = resolve_patient_id(raw_event["patient_id"], source_format)

    canonical_event = {
        "event_id":             raw_event.get("event_id") or str(uuid.uuid4()),
        "workforce_id":         workforce_id,
        "patient_id":           patient_id,
        "source_system":        source_format,
        "event_type":           normalize_event_type(raw_event["action_type"]),
        "resource_type":        normalize_resource_type(raw_event["resource"]),
        "action":               raw_event["action_type"],
        "observed_at":          raw_event["event_time"],
        "received_at":          datetime.now(timezone.utc).isoformat(),
        "device_id":            raw_event.get("workstation_id"),
        "application_context":  raw_event.get("application_screen"),
        "ip_address":           raw_event.get("source_ip"),
        "session_id":           raw_event.get("session_id"),
        "break_glass":          bool(raw_event.get("break_glass_override", False)),
        "break_glass_reason":   raw_event.get("break_glass_reason"),
    }

    kinesis.put_record(
        StreamName=AUDIT_EVENTS_STREAM,
        Data=json.dumps(canonical_event, default=str).encode("utf-8"),
        PartitionKey=workforce_id,   # partition by user for ordering within a session
    )

    # Persist to the raw event lake for retrospective analysis and
    # retraining. Partitioning by date and source keeps Athena and
    # Glue happy.
    obs_at = canonical_event["observed_at"]
    s3_client.put_object(
        Bucket=RAW_EVENTS_BUCKET,
        Key=(
            f"source={source_format}/year={obs_at[:4]}/"
            f"month={obs_at[5:7]}/day={obs_at[8:10]}/"
            f"{canonical_event['event_id']}.json"
        ),
        Body=json.dumps(canonical_event, default=str).encode("utf-8"),
        ServerSideEncryption="aws:kms",
    )

    return {"statusCode": 200, "event_id": canonical_event["event_id"]}

def is_off_hours(observed_at_iso, role):
    """Decide whether an access falls outside the user's typical working hours.

    A registered nurse on a normal day shift accessing at 3 a.m. is off-hours.
    A registered nurse on a night shift accessing at 3 a.m. is normal. The
    decision depends on the role's typical shift pattern. Production loads
    role-to-shift-window mappings from a versioned table.
    """
    obs_dt = datetime.fromisoformat(observed_at_iso.replace("Z", "+00:00"))
    hour = obs_dt.hour
    role_normal_windows = {
        "registered_nurse_day": (7, 19),
        "registered_nurse_night": (19, 7),
        "physician_day": (7, 19),
        "physician_night": (19, 7),
        "billing_analyst": (8, 18),
        "database_administrator": (8, 18),
        "default": (7, 19),
    }
    # Without per-user shift-pattern data, fall back to role default. The
    # cleaner production path is to use the actual scheduled-shift data
    # from Step 2's `scheduled_to_work` enrichment, which the rules
    # engine has access to.
    start, end = role_normal_windows.get(role, role_normal_windows["default"])
    if start < end:
        return not (start <= hour < end)
    # Wraps midnight (e.g., 19 to 7).
    return not (hour >= start or hour < end)

def is_off_shift(observed_at_iso, schedule):
    """Decide whether an access occurred while the user was off-shift."""
    if schedule is None:
        return True   # no scheduled shift on this date
    shift_start = schedule.get("shift_start")
    shift_end   = schedule.get("shift_end")
    if not shift_start or not shift_end:
        return False
    obs_dt = datetime.fromisoformat(observed_at_iso.replace("Z", "+00:00"))
    start_dt = datetime.fromisoformat(shift_start.replace("Z", "+00:00"))
    end_dt   = datetime.fromisoformat(shift_end.replace("Z", "+00:00"))
    return not (start_dt <= obs_dt <= end_dt)

def check_care_relationship(workforce_id, patient_id, as_of):
    """Look up documented care-relationship paths.

    Production queries Neptune for relationship paths through care_team,
    encounter, on_call, scheduling, order_signature, and documentation
    edges. The teaching example queries the in-process NetworkX graph
    that a later step builds. Returns a structured result with strength
    score (0..1).
    """
    graph = _global_graph()
    if graph is None:
        return {"has_any": False, "types": [], "strength_score": 0.0}

    # Direct path: workforce_id -> patient_id with up to 2 hops via care
    # relationship edges. Strength is graded by edge type.
    edge_strength = {
        "care_team":               1.0,
        "assigned_attending":      1.0,
        "assigned_nurse":          1.0,
        "on_call":                 0.9,
        "consult":                 0.85,
        "order_signature":         0.85,
        "documentation_authorship": 0.85,
        "scheduling":              0.7,
        "transitions_team":        0.65,
        "case_management":         0.65,
        "cross_coverage":          0.6,
        "team_membership":         0.5,
    }
    relationship_types_found = []
    best_strength = 0.0
    if graph.has_node(workforce_id) and graph.has_node(patient_id):
        # Direct edge
        if graph.has_edge(workforce_id, patient_id):
            edge_data = graph.get_edge_data(workforce_id, patient_id)
            edge_type = edge_data.get("type", "team_membership")
            relationship_types_found.append(edge_type)
            best_strength = max(best_strength,
                                edge_strength.get(edge_type, 0.5))
        # Two-hop path through encounter or department
        for neighbor in graph.neighbors(workforce_id):
            if graph.has_edge(neighbor, patient_id):
                first_hop = graph.get_edge_data(workforce_id, neighbor).get("type", "team_membership")
                second_hop = graph.get_edge_data(neighbor, patient_id).get("type", "team_membership")
                hop_strength = min(
                    edge_strength.get(first_hop, 0.4),
                    edge_strength.get(second_hop, 0.4),
                ) * 0.85   # decay across hops
                if hop_strength > best_strength:
                    best_strength = hop_strength
                    relationship_types_found.extend([first_hop, second_hop])

    return {
        "has_any":          len(relationship_types_found) > 0,
        "types":            list(set(relationship_types_found)),
        "strength_score":   best_strength,
    }

_GRAPH = None

def _global_graph():
    """Return the in-process relationship graph for the demo.

    Production reads from Amazon Neptune. The teaching example builds an
    in-process NetworkX graph at pipeline-init time so the relationship
    detector path is runnable without provisioning a Neptune cluster.
    """
    return _GRAPH

def geolocate_ip(ip_address):
    """Return a coarse geolocation label for an IP address.

    Production uses a maintained IP-to-geo service (MaxMind, IP2Location,
    or AWS Location Service) to identify country, region, and ISP. The
    teaching example returns a placeholder that distinguishes corporate
    vs external IPs by a CIDR prefix.
    """
    if ip_address is None:
        return {"network": "unknown", "country": "unknown"}
    if ip_address.startswith("10.") or ip_address.startswith("192.168."):
        return {"network": "corporate", "country": "us"}
    return {"network": "external", "country": "unknown"}

def is_unusual_for_user(workforce_id, geo):
    """Return True if the geo label is unusual for this user's history.

    Production queries Timestream or DynamoDB for the user's geo history
    and flags genuinely-new locations. The teaching example is a stub
    that flags any non-corporate network as unusual.
    """
    return geo.get("network") != "corporate"

def enrich_event(event):
    """Attach identity, schedule, care-team, and patient-flag enrichments."""
    identity_table = dynamodb.Table(WORKFORCE_IDENTITY_TABLE)
    identity_response = identity_table.get_item(
        Key={"workforce_id": event["workforce_id"]}
    )
    identity = _undecimalize(identity_response.get("Item")) or {}

    event["user_role"]         = identity.get("role")
    event["user_department"]   = identity.get("department")
    event["user_manager"]      = identity.get("manager_id")
    event["user_employment"]   = identity.get("employment_type")
    event["user_address_zip"]  = identity.get("address_zip")
    event["user_last_name"]    = (identity.get("last_name") or "").lower()
    event["user_hire_date"]    = identity.get("hire_date")

    days_since_hire = None
    if identity.get("hire_date"):
        hire_dt = datetime.fromisoformat(
            identity["hire_date"].replace("Z", "+00:00")
        )
        obs_dt = datetime.fromisoformat(
            event["observed_at"].replace("Z", "+00:00")
        )
        days_since_hire = (obs_dt - hire_dt).total_seconds() / 86400.0
    event["days_since_hire"] = days_since_hire

    # Workforce-link to patient identity. Some patients are also workforce
    # members (an employee who is a patient at the same facility); some
    # patients have a workforce-link via a household ID. Production
    # maintains both linkages with appropriate access controls.
    schedule_table = dynamodb.Table(WORKFORCE_SCHEDULE_TABLE)
    schedule_response = schedule_table.get_item(Key={
        "workforce_id": event["workforce_id"],
        "shift_date":   event["observed_at"][:10],
    })
    schedule = _undecimalize(schedule_response.get("Item"))
    event["scheduled_to_work"] = schedule is not None
    event["scheduled_unit"]    = schedule.get("unit") if schedule else None
    event["is_off_shift"]       = is_off_shift(event["observed_at"], schedule)
    event["is_off_hours"]       = is_off_hours(
        event["observed_at"],
        f"{event['user_role']}_day" if event.get("user_role") else "default",
    )

    # Care-relationship enrichment. The single most important enrichment
    # for distinguishing legitimate from problematic access.
    care = check_care_relationship(
        event["workforce_id"], event["patient_id"], event["observed_at"]
    )
    event["has_care_relationship"]      = care["has_any"]
    event["care_relationship_types"]    = care["types"]
    event["care_relationship_strength"] = care["strength_score"]

    # Patient context.
    patient_table = dynamodb.Table(PATIENT_CONTEXT_TABLE)
    patient_response = patient_table.get_item(
        Key={"patient_id": event["patient_id"]}
    )
    patient = _undecimalize(patient_response.get("Item")) or {}
    event["patient_sensitivity_flags"] = patient.get("sensitivity_flags", [])
    event["patient_is_employee"]       = bool(patient.get("is_workforce_member", False))
    event["patient_last_name"]         = (patient.get("last_name") or "").lower()
    event["patient_address_zip"]       = patient.get("address_zip")
    event["patient_household_id"]      = patient.get("household_id")
    event["patient_is_deceased"]       = bool(patient.get("is_deceased", False))
    event["patient_id_workforce_link"] = patient.get("workforce_id_link")

    # Network and device context.
    geo = geolocate_ip(event.get("ip_address"))
    event["geo_location"]    = geo
    event["is_off_network"]  = geo.get("network") != "corporate"
    event["is_unusual_geo"]  = is_unusual_for_user(event["workforce_id"], geo)

    return event

def compute_name_uniqueness(surname):
    """Return a 0..1 uniqueness score for a surname.

    Higher means rarer. Production uses a real surname-frequency reference
    (US Census or equivalent regional data); the demo uses the small
    SURNAME_UNIQUENESS table at the top of the file.
    """
    if not surname:
        return 0.0
    return SURNAME_UNIQUENESS.get(surname.lower(), SURNAME_UNIQUENESS["DEFAULT"])

def zip_population(zip_code):
    """Return population for a ZIP code (or a default if not in the table)."""
    if not zip_code:
        return ZIP_POPULATION["DEFAULT"]
    return ZIP_POPULATION.get(zip_code, ZIP_POPULATION["DEFAULT"])

def is_member_of_household(workforce_id, household_id):
    """Lookup whether a workforce member is in a household.

    Production maintains a workforce-to-household linkage table fed from
    HRIS dependent and address data, with appropriate access controls.
    The teaching example returns False; the household_id linkage path is
    documented but not exercised in the demo.
    """
    return False

def severity_from_break_glass_reason(reason_text):
    """Infer severity from the documented break-glass reason text.

    Specific clinical reasons get medium severity (the override is
    plausibly justified). Vague or absent reasons get high severity
    (the override may not be justified). Production uses a tuned
    classifier; the teaching example uses keyword heuristics.
    """
    if not reason_text:
        return "high"
    reason_lower = reason_text.lower()
    specific_keywords = [
        "code blue", "rapid response", "stroke alert", "trauma",
        "cardiac arrest", "consult requested by", "transfer of care",
        "covering for", "on-call coverage", "emergency consult",
    ]
    vague_keywords = [
        "review", "check", "look up", "verify", "see patient",
    ]
    if any(k in reason_lower for k in specific_keywords):
        return "medium"
    if any(k in reason_lower for k in vague_keywords):
        return "high"
    return "medium"

def run_rules_engine(event):
    """Evaluate the rule library against the enriched event.

    Returns a list of flag dicts with rule_id, severity, confidence, and
    a structured evidence payload. The composite scorer combines the
    flag list into a rules-confidence number and the case builder uses
    the evidence payloads in the investigator-facing display.
    """
    flags = []

    # RULE-001: Same last name with no documented care relationship.
    # Family-relationship access is the most common policy violation by
    # volume; same-name is the strongest single signal. Weighted by name
    # uniqueness so a Smith-on-Smith hit is much weaker than a
    # Wojnarowski-on-Wojnarowski hit.
    if (event.get("user_last_name")
            and event.get("patient_last_name")
            and event["user_last_name"] == event["patient_last_name"]
            and not event.get("has_care_relationship")):
        uniqueness = compute_name_uniqueness(event["user_last_name"])
        flags.append({
            "rule_id":     "RULE-001-SAME-LAST-NAME-NO-CARE",
            "severity":    "high",
            "confidence":  uniqueness,
            "evidence": {
                "user_last_name":     event["user_last_name"],
                "patient_last_name":  event["patient_last_name"],
                "name_uniqueness":    uniqueness,
            },
            "explanation": (
                f"User and patient share last name '{event['user_last_name']}' "
                f"(uniqueness {uniqueness:.2f}); no documented care "
                f"relationship found at access time."
            ),
        })

    # RULE-002 / RULE-003: Same household / same neighborhood.
    if (event.get("user_address_zip")
            and event.get("patient_address_zip")
            and event["user_address_zip"] == event["patient_address_zip"]):
        if (event.get("patient_household_id")
                and is_member_of_household(event["workforce_id"],
                                            event["patient_household_id"])):
            flags.append({
                "rule_id":    "RULE-002-SAME-HOUSEHOLD-NO-CARE",
                "severity":   "high",
                "confidence": 0.95,
                "evidence": {
                    "household_id": event["patient_household_id"],
                },
                "explanation": (
                    "User and patient share an HR-linked household; "
                    "no care relationship found."
                ),
            })
        elif (zip_population(event["user_address_zip"]) < SMALL_ZIP_THRESHOLD
              and not event.get("has_care_relationship")):
            flags.append({
                "rule_id":    "RULE-003-SAME-NEIGHBORHOOD-NO-CARE",
                "severity":   "medium",
                "confidence": 0.65,
                "evidence": {
                    "zip":              event["user_address_zip"],
                    "zip_population":   zip_population(event["user_address_zip"]),
                },
                "explanation": (
                    f"User home ZIP and patient home ZIP both "
                    f"{event['user_address_zip']} (small ZIP, "
                    f"{zip_population(event['user_address_zip'])} residents); "
                    f"no care relationship."
                ),
            })

    # RULE-010: Self-access. Severity is policy-dependent: some orgs allow
    # self-access for accessing one's own records, some forbid it.
    if (event.get("patient_id_workforce_link")
            and event["patient_id_workforce_link"] == event["workforce_id"]):
        flags.append({
            "rule_id":    "RULE-010-SELF-ACCESS",
            "severity":   "policy_dependent",
            "confidence": 1.0,
            "evidence": {
                "workforce_id":  event["workforce_id"],
                "patient_id":    event["patient_id"],
            },
            "explanation": (
                "User is accessing their own patient record. "
                "Severity depends on org policy."
            ),
        })

    # RULE-020: Sensitive-patient access without strong care relationship.
    if ("VIP" in (event.get("patient_sensitivity_flags") or [])
            and event.get("care_relationship_strength", 0.0) < 0.8):
        flags.append({
            "rule_id":    "RULE-020-VIP-WEAK-CARE",
            "severity":   "high",
            "confidence": 0.85,
            "evidence": {
                "sensitivity_flags": event["patient_sensitivity_flags"],
                "care_strength":      event.get("care_relationship_strength"),
            },
            "explanation": (
                "Access to VIP-flagged patient without a strong care "
                "relationship."
            ),
        })

    # RULE-021: Co-worker access without strong care relationship.
    if (event.get("patient_is_employee")
            and event.get("care_relationship_strength", 0.0) < 0.8):
        flags.append({
            "rule_id":    "RULE-021-EMPLOYEE-PATIENT-WEAK-CARE",
            "severity":   "high",
            "confidence": 0.80,
            "evidence": {
                "care_strength": event.get("care_relationship_strength"),
            },
            "explanation": (
                "Access to a workforce-member patient without a strong "
                "care relationship."
            ),
        })

    # RULE-030: Break-glass override. Always flagged for review; severity
    # tuned by reason quality.
    if event.get("break_glass"):
        severity = severity_from_break_glass_reason(
            event.get("break_glass_reason")
        )
        flags.append({
            "rule_id":    "RULE-030-BREAK-GLASS-OVERRIDE",
            "severity":   severity,
            "confidence": 1.0,
            "evidence": {
                "reason":             event.get("break_glass_reason"),
                "care_relationship":  event.get("has_care_relationship"),
                "sensitivity_flags":  event.get("patient_sensitivity_flags"),
            },
            "explanation": (
                f"Break-glass override used. Reason: "
                f"'{event.get('break_glass_reason') or '(none provided)'}'."
            ),
        })

    # RULE-040: Off-hours access by users on standard daytime schedules
    # without scheduled coverage.
    if (event.get("is_off_hours")
            and not event.get("scheduled_to_work")
            and event.get("user_employment") in ("employee", "contractor")):
        flags.append({
            "rule_id":    "RULE-040-OFF-HOURS-NO-SCHEDULE",
            "severity":   "low",
            "confidence": 0.50,
            "evidence": {
                "observed_at":  event["observed_at"],
                "user_role":    event.get("user_role"),
            },
            "explanation": (
                "Access outside the user's role-typical hours and no "
                "scheduled shift on this date."
            ),
        })

    # RULE-050: Deceased patient access without care relationship.
    if (event.get("patient_is_deceased")
            and not event.get("has_care_relationship")):
        flags.append({
            "rule_id":    "RULE-050-DECEASED-PATIENT-NO-CARE",
            "severity":   "medium",
            "confidence": 0.70,
            "evidence": {},
            "explanation": (
                "Access to a deceased patient's record without a "
                "documented care relationship."
            ),
        })

    # RULE-060: Print or export with weak care relationship or sensitivity
    # flags.
    if (event.get("event_type") in ("print", "export")
            and (event.get("patient_sensitivity_flags")
                 or event.get("care_relationship_strength", 0.0) < 0.5)):
        flags.append({
            "rule_id":    "RULE-060-EXPORT-WEAK-CARE",
            "severity":   "medium",
            "confidence": 0.70,
            "evidence": {
                "event_type":         event["event_type"],
                "sensitivity_flags":  event.get("patient_sensitivity_flags"),
                "care_strength":      event.get("care_relationship_strength"),
            },
            "explanation": (
                f"{event['event_type'].title()} action on a record with "
                f"weak care relationship or sensitivity flags."
            ),
        })

    return flags

def max_severity_confidence(flags):
    """Combine a flag list into a single rules-confidence score."""
    if not flags:
        return 0.0
    # Take the maximum confidence weighted by severity floor. A high-
    # severity rule with confidence 0.81 beats a low-severity rule with
    # confidence 1.0.
    max_score = 0.0
    for f in flags:
        floor = RULE_SEVERITY_FLOOR.get(f["severity"], 0.5)
        score = max(f.get("confidence", 0.0), floor)
        if score > max_score:
            max_score = score
    return max_score

def derive_peer_group(role, department, shift_pattern):
    """Compose a peer-group identifier from role, department, and shift.

    Peer-group definition is one of the most consequential design choices
    in the entire system. Bad peer groups produce bad baselines and bad
    alerts. Production tunes peer groups against historical adjudicated
    cases; this stub uses a simple concatenation.
    """
    return f"{role or 'unknown'}|{department or 'unknown'}|{shift_pattern or 'unknown'}"

def aggregate_user_activity(workforce_id, window_hours, ending_at):
    """Aggregate a user's recent activity over a window.

    Production reads from Timestream for the per-user time-series of
    counts, plus DynamoDB for the per-user state record that holds the
    rolling counters. The teaching example reads from the user-state
    table only and synthesizes the per-window aggregates from a small
    in-memory event window.
    """
    table = dynamodb.Table(USER_STATE_TABLE)
    response = table.get_item(Key={"workforce_id": workforce_id})
    state = _undecimalize(response.get("Item")) or {}

    recent_events = state.get("recent_events", [])
    end_dt = datetime.fromisoformat(ending_at.replace("Z", "+00:00"))
    start_dt = end_dt - timedelta(hours=window_hours)

    in_window = []
    for e in recent_events:
        e_dt = datetime.fromisoformat(e["observed_at"].replace("Z", "+00:00"))
        if start_dt <= e_dt <= end_dt:
            in_window.append(e)

    if not in_window:
        return {
            "event_count":                       0,
            "unique_patients":                   0,
            "unique_resource_types":             0,
            "export_count":                      0,
            "print_count":                       0,
            "break_glass_count":                 0,
            "off_hours_fraction":                0.0,
            "never_seen_before_fraction":        0.0,
            "sensitive_patient_fraction":        0.0,
            "weak_care_relationship_fraction":   0.0,
        }

    unique_patients = {e["patient_id"] for e in in_window}
    unique_resources = {e.get("resource_type") for e in in_window if e.get("resource_type")}
    seen_before = set(state.get("known_patients", []))
    sensitive = sum(1 for e in in_window if e.get("patient_sensitivity_flags"))
    weak_care = sum(1 for e in in_window
                    if e.get("care_relationship_strength", 1.0) < 0.5)
    off_hours = sum(1 for e in in_window if e.get("is_off_hours"))

    return {
        "event_count":                     len(in_window),
        "unique_patients":                 len(unique_patients),
        "unique_resource_types":           len(unique_resources),
        "export_count":                    sum(1 for e in in_window if e.get("event_type") == "export"),
        "print_count":                     sum(1 for e in in_window if e.get("event_type") == "print"),
        "break_glass_count":               sum(1 for e in in_window if e.get("break_glass")),
        "off_hours_fraction":              off_hours / len(in_window),
        "never_seen_before_fraction":      sum(1 for p in unique_patients if p not in seen_before) / max(len(unique_patients), 1),
        "sensitive_patient_fraction":      sensitive / len(in_window),
        "weak_care_relationship_fraction": weak_care / len(in_window),
    }

def get_user_baseline(workforce_id):
    """Load the per-user historical baseline from the user-state store.

    Production stores the baseline as a rolling-window summary refreshed
    nightly (or on a faster cadence for very active users). The summary
    holds the mean and standard deviation per feature, plus the first-
    observed timestamp for cold-start handling.
    """
    table = dynamodb.Table(USER_STATE_TABLE)
    response = table.get_item(Key={"workforce_id": workforce_id})
    state = _undecimalize(response.get("Item")) or {}
    baseline = state.get("behavioral_baseline") or {}
    return {
        "mean":              baseline.get("mean", {}),
        "std":               baseline.get("std", {}),
        "first_observed":    baseline.get("first_observed"),
        "shift_pattern":     state.get("shift_pattern", "unknown"),
    }

def get_peer_baseline(peer_group_id):
    """Load the peer-group baseline.

    Production maintains peer-group baselines in a versioned reference
    table refreshed nightly from the aggregated user-state data. The
    teaching example returns a synthetic baseline.
    """
    table = dynamodb.Table(USER_STATE_TABLE)
    response = table.get_item(Key={"workforce_id": f"PEER_BASELINE::{peer_group_id}"})
    state = _undecimalize(response.get("Item")) or {}
    baseline = state.get("behavioral_baseline") or {}
    return {
        "mean":  baseline.get("mean", {}),
        "std":   baseline.get("std", {}),
    }

# Feature weights tune which deviations matter most. Export volume and
# sensitive-patient fraction are weighted higher because they represent
# higher-stakes patterns; off-hours fraction is weighted lower because
# it has higher false-positive rate in clinical workforce.
FEATURE_WEIGHTS = {
    "events_count_24_hour":             1.0,
    "events_count_1_hour":              0.8,
    "unique_patients_24_hour":          1.0,
    "unique_patients_7_day":            0.8,
    "export_count_24_hour":             2.0,
    "export_count_7_day":               2.0,
    "print_count_24_hour":              1.5,
    "break_glass_count_7_day":          1.5,
    "off_hours_fraction_24_hour":       0.6,
    "new_patient_fraction_24_hour":     1.2,
    "sensitive_patient_fraction_24_hour": 1.8,
    "weak_care_fraction_24_hour":       1.5,
}

def weighted_max_z(per_feature_z, weights):
    """Compute a weighted aggregate of feature z-scores.

    Max-of-z catches the strongest single deviation; the per-feature
    weight skews the aggregate toward the higher-stakes features. A
    weighted Mahalanobis distance would be more theoretically clean
    but max-of-z works well in practice and explains better.
    """
    weighted = []
    for feature, z in per_feature_z.items():
        w = weights.get(feature, 1.0)
        weighted.append(abs(z) * w)
    return max(weighted) if weighted else 0.0

def days_since(timestamp_iso):
    """Compute days since an ISO-8601 timestamp."""
    if timestamp_iso is None:
        return 0.0
    obs_dt = datetime.fromisoformat(timestamp_iso.replace("Z", "+00:00"))
    now_dt = datetime.now(timezone.utc)
    return (now_dt - obs_dt).total_seconds() / 86400.0

def run_baseline_detector(event):
    """Compute per-user and peer-group deviation scores.

    Multiple windows in parallel catch fast and slow shifts. The weighted
    max-of-z aggregate gives a single deviation score that the composite
    layer consumes. SHAP-style feature attribution is preserved in the
    per-feature z-score dict for the explanation layer.
    """
    workforce_id = event["workforce_id"]
    feature_vector = {}

    # Aggregate per-window features. Production runs these queries against
    # Timestream and DynamoDB; the teaching example uses an in-memory
    # event window.
    windows = [(1, "1_hour"), (8, "8_hour"), (24, "24_hour"),
               (168, "7_day"), (720, "30_day")]
    for window_hours, suffix in windows:
        agg = aggregate_user_activity(workforce_id, window_hours, event["observed_at"])
        feature_vector[f"events_count_{suffix}"]              = agg["event_count"]
        feature_vector[f"unique_patients_{suffix}"]            = agg["unique_patients"]
        feature_vector[f"unique_resources_{suffix}"]           = agg["unique_resource_types"]
        feature_vector[f"export_count_{suffix}"]               = agg["export_count"]
        feature_vector[f"print_count_{suffix}"]                = agg["print_count"]
        feature_vector[f"break_glass_count_{suffix}"]          = agg["break_glass_count"]
        feature_vector[f"off_hours_fraction_{suffix}"]         = agg["off_hours_fraction"]
        feature_vector[f"new_patient_fraction_{suffix}"]       = agg["never_seen_before_fraction"]
        feature_vector[f"sensitive_patient_fraction_{suffix}"]  = agg["sensitive_patient_fraction"]
        feature_vector[f"weak_care_fraction_{suffix}"]         = agg["weak_care_relationship_fraction"]

    # Patient-specific baseline.
    user_baseline = get_user_baseline(workforce_id)
    per_feature_z = {}
    for feature_name, feature_value in feature_vector.items():
        baseline_mean = user_baseline["mean"].get(feature_name, feature_value)
        baseline_std  = user_baseline["std"].get(feature_name, 1.0) or 1.0
        per_feature_z[feature_name] = (feature_value - baseline_mean) / max(baseline_std, 1e-6)

    # Peer-group baseline.
    peer_group_id = derive_peer_group(
        event.get("user_role"),
        event.get("user_department"),
        user_baseline["shift_pattern"],
    )
    peer_baseline = get_peer_baseline(peer_group_id)
    per_feature_peer_z = {}
    for feature_name, feature_value in feature_vector.items():
        peer_mean = peer_baseline["mean"].get(feature_name, feature_value)
        peer_std  = peer_baseline["std"].get(feature_name, 1.0) or 1.0
        per_feature_peer_z[feature_name] = (feature_value - peer_mean) / max(peer_std, 1e-6)

    composite_user_z = weighted_max_z(per_feature_z, FEATURE_WEIGHTS)
    composite_peer_z = weighted_max_z(per_feature_peer_z, FEATURE_WEIGHTS)

    # Cold-start handling: new users with insufficient history fall back
    # to peer-group comparison only.
    baseline_age_days = days_since(user_baseline.get("first_observed"))
    if baseline_age_days < MIN_BASELINE_DAYS:
        composite = composite_peer_z
        baseline_source = "peer_only_cold_start"
    else:
        composite = max(composite_user_z, composite_peer_z)
        baseline_source = "patient_specific_and_peer"

    # Sigmoid maps the unbounded weighted-z to a 0..1 score.
    deviation_score = 1.0 / (1.0 + math.exp(-composite / 3.0))

    return {
        "deviation_score":     deviation_score,
        "per_feature_z":       per_feature_z,
        "per_feature_peer_z":  per_feature_peer_z,
        "peer_group_id":       peer_group_id,
        "baseline_source":     baseline_source,
        "baseline_age_days":   baseline_age_days,
        "feature_snapshot":    feature_vector,
    }

def query_neptune_for_paths(workforce_id, patient_id, max_hops=3):
    """Query Amazon Neptune for relationship paths between user and patient.

    Production sends Gremlin to Neptune. The teaching example queries
    the in-process NetworkX graph. This function is structured to make
    the production swap straightforward: replace the body with a Neptune
    Gremlin call and the rest of the detector works unchanged.

    Production Gremlin (illustrative):
        g.V().has('workforce', 'workforce_id', workforce_id)
          .repeat(both('care_team', 'on_call', 'consult', 'scheduling',
                       'order_signature', 'documentation_authorship',
                       'team_membership', 'cross_coverage'))
          .times(max_hops)
          .has('patient', 'patient_id', patient_id)
          .path()
          .limit(5)
    """
    graph = _global_graph()
    if graph is None or not graph.has_node(workforce_id) or not graph.has_node(patient_id):
        return []

    paths = []
    try:
        # nx.all_simple_paths is fine for the small demo graph; a real
        # Neptune query is what scales.
        for p in nx.all_simple_paths(graph, source=workforce_id,
                                      target=patient_id, cutoff=max_hops):
            paths.append({"length": len(p) - 1, "nodes": p})
            if len(paths) >= 5:
                break
    except nx.NetworkXNoPath:
        pass
    return paths

def query_unit_overlap(workforce_id, patient_id, observed_at, lookback_days=7):
    """Check whether the patient was on a unit that the user covers.

    Department-level relationship: was the patient on a unit this user's
    department covers, even if the user wasn't directly assigned? Captures
    floor-coverage, cross-coverage, and pre-admission-prep patterns that
    the EHR's care-team module misses.
    """
    graph = _global_graph()
    if graph is None:
        return False

    user_dept = graph.nodes[workforce_id].get("department") if graph.has_node(workforce_id) else None
    if not user_dept:
        return False

    # Walk patient's recent unit assignments and check department overlap.
    if not graph.has_node(patient_id):
        return False
    for neighbor in graph.neighbors(patient_id):
        edge = graph.get_edge_data(patient_id, neighbor)
        if edge.get("type") == "admitted_to_unit":
            unit_dept = graph.nodes[neighbor].get("department")
            if unit_dept == user_dept:
                return True
    return False

def get_user_recent_patient_set(workforce_id, hours=24):
    """Return the set of patients this user has accessed in the last N hours."""
    table = dynamodb.Table(USER_STATE_TABLE)
    response = table.get_item(Key={"workforce_id": workforce_id})
    state = _undecimalize(response.get("Item")) or {}
    end_dt = datetime.now(timezone.utc)
    start_dt = end_dt - timedelta(hours=hours)
    patients = set()
    for e in state.get("recent_events", []):
        e_dt = datetime.fromisoformat(e["observed_at"].replace("Z", "+00:00"))
        if start_dt <= e_dt <= end_dt:
            patients.add(e["patient_id"])
    return patients

def compute_graph_cohesion(patient_set):
    """Return a 0..1 cohesion score for a set of patients.

    High cohesion means the patients share care teams, units, or
    diagnosis groups (a clinical workflow). Low cohesion means the
    patients are scattered with no shared structural connection
    (a credential-compromise reconnaissance pattern).
    """
    graph = _global_graph()
    if graph is None or len(patient_set) < 2:
        return 1.0   # too small to evaluate; assume cohesive

    # Count how many pairs share at least one unit or care-team neighbor.
    patient_list = list(patient_set)
    cohesive_pairs = 0
    total_pairs = 0
    for i in range(len(patient_list)):
        for j in range(i + 1, len(patient_list)):
            total_pairs += 1
            p1, p2 = patient_list[i], patient_list[j]
            if not (graph.has_node(p1) and graph.has_node(p2)):
                continue
            shared = set(graph.neighbors(p1)) & set(graph.neighbors(p2))
            if shared:
                cohesive_pairs += 1
    if total_pairs == 0:
        return 1.0
    return cohesive_pairs / total_pairs

def check_family_link(workforce_id, patient_id):
    """Look up an HR-linked household connection between user and patient.

    Production checks the workforce-to-household linkage table fed from
    HRIS dependent and address data. Returns True for documented family
    relationships (spouse, dependent, declared dependent contact). The
    teaching example returns False; the linkage path is documented but
    not exercised in the demo.
    """
    return False

CLUSTER_THRESHOLD = 8
CLUSTER_COHESION_THRESHOLD = 0.3

def run_graph_detector(event):
    """Evaluate the relationship-graph detector for a single event."""
    paths = query_neptune_for_paths(
        event["workforce_id"], event["patient_id"], max_hops=3
    )
    has_direct = any(p["length"] <= 2 for p in paths)
    has_indirect = any(p["length"] <= 4 for p in paths)

    department_overlap = query_unit_overlap(
        event["workforce_id"], event["patient_id"], event["observed_at"]
    )

    recent_patients = get_user_recent_patient_set(event["workforce_id"])
    cluster_cohesion = None
    cluster_anomaly = False
    if len(recent_patients) >= CLUSTER_THRESHOLD:
        cluster_cohesion = compute_graph_cohesion(recent_patients)
        cluster_anomaly = cluster_cohesion < CLUSTER_COHESION_THRESHOLD

    family_match = check_family_link(event["workforce_id"], event["patient_id"])

    relationship_evidence = {
        "has_direct_relationship":    has_direct,
        "has_indirect_relationship":  has_indirect,
        "department_overlap":         department_overlap,
        "cluster_cohesion_score":     cluster_cohesion,
        "cluster_anomaly":            cluster_anomaly,
        "family_match":               family_match,
        "path_count":                 len(paths),
    }

    if has_direct:
        graph_score = 0.05
    elif has_indirect or department_overlap:
        graph_score = 0.30
    elif family_match:
        graph_score = 0.95
    elif cluster_anomaly:
        graph_score = 0.85
    else:
        graph_score = 0.65   # no documented connection at all

    return {
        "graph_score":            graph_score,
        "relationship_evidence":  relationship_evidence,
    }

def cohort_for_user(role, days_since_hire):
    """Map a user to a scoring cohort.

    The cohort drives detector weighting and tier thresholds. Privileged
    users get different weights than clinical workforce; new users (under
    90 days since hire) get reduced rules weight because their
    discovery-and-search behavior trips baselines for learning reasons.
    """
    if role in PRIVILEGED_ROLES:
        return "privileged_user"
    if days_since_hire is not None and days_since_hire < 90:
        return "new_user"
    if role in CLINICAL_ROLES:
        return "clinical_workforce"
    return "DEFAULT"

def cohort_weights_for(role, days_since_hire):
    """Return the detector weights for a user's cohort."""
    cohort = cohort_for_user(role, days_since_hire)
    return DEFAULT_DETECTOR_WEIGHTS.get(cohort, DEFAULT_DETECTOR_WEIGHTS["DEFAULT"])

def apply_calibration(raw_score, calibrator, subgroup):
    """Apply a (subgroup-aware) calibration mapping.

    Production maintains per-subgroup calibration curves (per role, per
    department) when calibration drift differs across cohorts. The
    teaching example uses a single isotonic regressor and ignores
    subgroup; the function signature shows the production shape.
    """
    if calibrator is None:
        return raw_score
    return float(calibrator.predict(np.array([raw_score]))[0])

def tier_from_score(score, role, days_since_hire):
    """Map a calibrated score to a privacy-office tier."""
    cohort = cohort_for_user(role, days_since_hire)
    thresholds = DEFAULT_TIER_THRESHOLDS.get(cohort, DEFAULT_TIER_THRESHOLDS["DEFAULT"])
    if score >= thresholds["tier_1"]:
        return "tier_1"
    if score >= thresholds["tier_2"]:
        return "tier_2"
    if score >= thresholds["tier_3"]:
        return "tier_3"
    return "below_threshold"

def composite_score(event, rules_flags, baseline_output, graph_output,
                    sequence_output, calibrator):
    """Combine detector outputs into a single calibrated case score."""
    weights = cohort_weights_for(event.get("user_role"),
                                 event.get("days_since_hire"))
    rules_confidence = max_severity_confidence(rules_flags)
    sequence_score   = sequence_output.get("sequence_score", 0.0)

    raw_composite = (
        weights["rules"]    * rules_confidence
      + weights["baseline"] * baseline_output["deviation_score"]
      + weights["graph"]    * graph_output["graph_score"]
      + weights["sequence"] * sequence_score
    )

    calibrated = apply_calibration(
        raw_composite, calibrator,
        subgroup=cohort_for_user(event.get("user_role"),
                                 event.get("days_since_hire")),
    )
    tier = tier_from_score(
        calibrated, event.get("user_role"), event.get("days_since_hire")
    )

    return {
        "score_id":                generate_score_id(),
        "event_id":                event["event_id"],
        "workforce_id":            event["workforce_id"],
        "patient_id":              event["patient_id"],
        "scored_at":               datetime.now(timezone.utc).isoformat(),
        "rules_flags":             rules_flags,
        "rules_confidence":        rules_confidence,
        "baseline_deviation_score": baseline_output["deviation_score"],
        "graph_score":             graph_output["graph_score"],
        "sequence_score":          sequence_score,
        "composite_raw":           raw_composite,
        "composite_calibrated":    calibrated,
        "tier":                    tier,
        "per_feature_z":           baseline_output["per_feature_z"],
        "relationship_evidence":   graph_output["relationship_evidence"],
        "feature_snapshot":        baseline_output["feature_snapshot"],
        "model_version":           "access-anomaly-composite-v3.1",
        "calibration_version":     "calib-v3.1-2026-04",
        "cohort":                  cohort_for_user(event.get("user_role"),
                                                    event.get("days_since_hire")),
    }

def generate_score_id():
    """Generate a score identifier."""
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    return f"SCORE-{today}-{uuid.uuid4().hex[:8]}"

def score_via_sagemaker_endpoint(features, feature_order):
    """Invoke the deployed SageMaker endpoint for the composite anomaly model.

    Production typically uses SageMaker batch transform for the nightly
    user-graph rescoring (cheaper at this cadence) and a real-time
    endpoint for event-driven re-scoring on high-priority triggers.
    Both accept the same feature payload; the difference is how
    invocation is triggered.
    """
    payload = ",".join(
        str(features.get(f, 0.0)) for f in feature_order
    )
    response = sagemaker_runtime.invoke_endpoint(
        EndpointName=SAGEMAKER_ENDPOINT_NAME,
        ContentType="text/csv",
        Body=payload,
    )
    body = response["Body"].read().decode("utf-8").strip()
    return float(body.split(",")[0])

def find_existing_case(workforce_id, patient_id, within_window_days):
    """Search for an open case grouping this user-patient pair.

    Production indexes case_state by (workforce_id, patient_id) for this
    lookup. The teaching example does a small in-memory scan.
    """
    table = dynamodb.Table(CASE_STATE_TABLE)
    cutoff = (datetime.now(timezone.utc)
              - timedelta(days=within_window_days)).isoformat()
    # In production, query via a GSI on (workforce_id, patient_id, opened_at).
    response = table.scan(
        FilterExpression=(
            (Key("workforce_id").eq(workforce_id))
            & (Key("patient_id").eq(patient_id))
            & (Key("status").eq("open_for_review"))
            & (Key("opened_at").gte(cutoff))
        ),
        Limit=10,
    )
    items = response.get("Items", [])
    return _undecimalize(items[0]) if items else None

def update_existing_case(existing_case, score_record):
    """Append a new score to an existing case and refresh evidence."""
    table = dynamodb.Table(CASE_STATE_TABLE)
    score_ids = existing_case.get("scoring_record_ids", []) + [score_record["score_id"]]
    composite = max(
        existing_case.get("composite_score", 0.0),
        float(score_record["composite_calibrated"]),
    )
    table.update_item(
        Key={"case_id": existing_case["case_id"]},
        UpdateExpression="SET scoring_record_ids = :s, composite_score = :c, updated_at = :t",
        ExpressionAttributeValues={
            ":s": score_ids,
            ":c": _to_decimal(composite),
            ":t": datetime.now(timezone.utc).isoformat(),
        },
    )

def check_recent_dismissal(score_record):
    """Was this same pattern recently dismissed for this user?"""
    table = dynamodb.Table(SUPPRESSION_RULES_TABLE)
    response = table.query(
        KeyConditionExpression=Key("workforce_id").eq(score_record["workforce_id"]),
    )
    for item in response.get("Items", []):
        rule = _undecimalize(item)
        valid_until = rule.get("valid_until")
        if valid_until and valid_until > datetime.now(timezone.utc).isoformat():
            # Match if the rule scope includes the patient or the patient
            # set this score touches.
            if rule.get("patient_id") == score_record["patient_id"]:
                return True
            if rule.get("scope") == "all_patients":
                return True
    return False

def add_suppression_rule(workforce_id, patient_id, reason, valid_for_days):
    """Add a suppression rule so a recently-dismissed pattern doesn't re-flag."""
    table = dynamodb.Table(SUPPRESSION_RULES_TABLE)
    valid_until_dt = datetime.now(timezone.utc) + timedelta(days=valid_for_days)
    rule = {
        "workforce_id":     workforce_id,
        "rule_id":          str(uuid.uuid4()),
        "patient_id":       patient_id,
        "scope":            "patient_specific",
        "reason":            reason,
        "added_at":         datetime.now(timezone.utc).isoformat(),
        "valid_until":       valid_until_dt.isoformat(),
        "ttl":               int(valid_until_dt.timestamp()),   # DynamoDB TTL
    }
    table.put_item(Item=_decimalize(rule))

def fetch_workforce_record(workforce_id):
    """Return the full workforce identity record for case display."""
    table = dynamodb.Table(WORKFORCE_IDENTITY_TABLE)
    response = table.get_item(Key={"workforce_id": workforce_id})
    return _undecimalize(response.get("Item")) or {}

def fetch_patient_record(patient_id):
    """Return the full patient context record for case display."""
    table = dynamodb.Table(PATIENT_CONTEXT_TABLE)
    response = table.get_item(Key={"patient_id": patient_id})
    return _undecimalize(response.get("Item")) or {}

def fetch_recent_user_activity(workforce_id, days):
    """Return a summary of the user's recent activity."""
    table = dynamodb.Table(USER_STATE_TABLE)
    response = table.get_item(Key={"workforce_id": workforce_id})
    state = _undecimalize(response.get("Item")) or {}
    cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    recent = [e for e in state.get("recent_events", [])
              if e.get("observed_at", "") >= cutoff]
    return {
        "event_count":           len(recent),
        "unique_patients":       len({e["patient_id"] for e in recent}),
        "export_count":           sum(1 for e in recent if e.get("event_type") == "export"),
        "break_glass_count":      sum(1 for e in recent if e.get("break_glass")),
        "off_hours_count":        sum(1 for e in recent if e.get("is_off_hours")),
    }

def top_n_drivers(per_feature_z, n=5):
    """Return the top-N features by absolute z-score."""
    items = [(name, z) for name, z in per_feature_z.items() if z is not None]
    items.sort(key=lambda x: abs(x[1]), reverse=True)
    return [{"feature": name, "z_score": float(z)} for name, z in items[:n]]

def build_case_narrative_prompt(evidence):
    """Build the Bedrock prompt for case-narrative generation.

    Constrained: cite the evidence, describe the access pattern, never
    assert intent or recommend an outcome. Always with human review;
    the LLM produces decision support, not decisions.
    """
    triggering = evidence.get("triggering_events", [])
    if triggering:
        first = triggering[0]
        access_summary = (
            f"At {first.get('observed_at')}, workforce member "
            f"{evidence['workforce_record'].get('workforce_id')} "
            f"({evidence['workforce_record'].get('role') or 'unknown role'}, "
            f"{evidence['workforce_record'].get('department') or 'unknown department'}) "
            f"accessed the chart of patient "
            f"{evidence['patient_record'].get('patient_id')}."
        )
    else:
        access_summary = "Multiple access events flagged in this case."

    rules_summary = "\n".join(
        f"- {f.get('explanation', f.get('rule_id'))}"
        for f in evidence.get("rules_flags", [])
    ) or "(no rules-engine flags)"

    care_summary = (
        f"Care relationship at access time: "
        f"{'documented' if evidence['care_relationship'].get('has_direct_relationship') else 'not documented'}; "
        f"department overlap: "
        f"{evidence['care_relationship'].get('department_overlap')}; "
        f"path count: {evidence['care_relationship'].get('path_count', 0)}."
    )

    baseline_summary = (
        f"Baseline deviation score: "
        f"{evidence['baseline_evidence']['deviation_score']:.2f}. "
        f"Top deviation features: "
        + ", ".join(d["feature"] for d in evidence["baseline_evidence"]["top_z_features"])
    )

    return (
        "You are summarizing a healthcare access-anomaly case for a "
        "privacy-office investigator. You are not making a determination "
        "of policy violation and you are not asserting intent. You are "
        "translating the structured evidence into a 2-4 sentence "
        "investigator-readable narrative that cites the evidence and "
        "describes the access pattern. End with the phrase 'This is "
        "decision support; investigator judgment governs.'\n\n"
        f"Composite score: {evidence['composite_score']:.2f} (tier {evidence['tier']})\n"
        f"{access_summary}\n\n"
        f"Rules-engine flags:\n{rules_summary}\n\n"
        f"{care_summary}\n\n"
        f"{baseline_summary}\n\n"
        "Produce 2-4 sentences of plain narrative. No bullet points. "
        "No assertions of intent. No recommended employment actions."
    )

def invoke_bedrock_narrative(evidence):
    """Generate the investigator-facing case narrative via Bedrock.

    Confirm the chosen Bedrock model is HIPAA-eligible under your AWS
    BAA before deploying.
    """
    prompt = build_case_narrative_prompt(evidence)
    try:
        response = bedrock_runtime.invoke_model(
            modelId=BEDROCK_MODEL_ID,
            body=json.dumps({
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens":        500,
                "temperature":       0.0,
                "messages":          [{"role": "user", "content": prompt}],
            }),
        )
        body = json.loads(response["body"].read())
        return body["content"][0]["text"].strip()
    except Exception as e:
        logger.warning("bedrock invocation failed", extra={"error": str(e)})
        _emit_metric("BedrockNarrativeFailed", 1)
        # Fall back to a structured-only summary so case-builder doesn't fail.
        return (
            f"Tier {evidence['tier']} case with composite score "
            f"{evidence['composite_score']:.2f}. "
            f"Rules flags: {len(evidence.get('rules_flags', []))}. "
            f"Care relationship: "
            f"{'documented' if evidence['care_relationship'].get('has_direct_relationship') else 'not documented'}. "
            "This is decision support; investigator judgment governs."
        )

def _emit_metric(metric_name, value, unit="Count"):
    """Emit a CloudWatch metric for operational monitoring."""
    try:
        cloudwatch.put_metric_data(
            Namespace="AccessAnomaly/CaseBuilder",
            MetricData=[{
                "MetricName": metric_name,
                "Value":      float(value),
                "Unit":       unit,
                "Timestamp":  datetime.now(timezone.utc),
            }],
        )
    except Exception as e:
        logger.warning("metric emit failed",
                       extra={"metric": metric_name, "error": str(e)})

def build_case(score_record, enriched_event):
    """Group, suppress, assemble evidence, and persist a case."""
    if score_record["tier"] == "below_threshold":
        return None

    existing_case = find_existing_case(
        score_record["workforce_id"], score_record["patient_id"],
        within_window_days=CASE_GROUPING_WINDOW_DAYS,
    )
    if existing_case:
        update_existing_case(existing_case, score_record)
        return existing_case["case_id"]

    if check_recent_dismissal(score_record):
        logger.info("case suppressed by recent dismissal",
                    extra={"workforce_id": score_record["workforce_id"],
                           "patient_id":   score_record["patient_id"]})
        _emit_metric("Suppressed_RecentDismissal", 1)
        return None

    workforce_record = fetch_workforce_record(score_record["workforce_id"])
    patient_record   = fetch_patient_record(score_record["patient_id"])
    recent_activity  = fetch_recent_user_activity(score_record["workforce_id"], days=30)

    evidence = {
        "triggering_events":  [enriched_event],
        "workforce_record":   {
            "workforce_id":      score_record["workforce_id"],
            "role":              workforce_record.get("role"),
            "department":        workforce_record.get("department"),
            "manager":           workforce_record.get("manager_id"),
            "employment_type":   workforce_record.get("employment_type"),
            "hire_date":         workforce_record.get("hire_date"),
        },
        "patient_record": {
            "patient_id":         score_record["patient_id"],
            "sensitivity_flags":  patient_record.get("sensitivity_flags", []),
            "is_workforce_member": patient_record.get("is_workforce_member", False),
            "is_deceased":         patient_record.get("is_deceased", False),
        },
        "care_relationship":  score_record["relationship_evidence"],
        "user_recent_activity": recent_activity,
        "rules_flags":        score_record["rules_flags"],
        "baseline_evidence": {
            "deviation_score":  score_record["baseline_deviation_score"],
            "top_z_features":   top_n_drivers(score_record["per_feature_z"], n=5),
        },
        "graph_evidence":     score_record["relationship_evidence"],
        "composite_score":    float(score_record["composite_calibrated"]),
        "tier":               score_record["tier"],
        "model_version":      score_record["model_version"],
        "calibration_version": score_record["calibration_version"],
    }

    evidence["narrative"] = invoke_bedrock_narrative(evidence)

    case = {
        "case_id":            f"CASE-{datetime.now(timezone.utc).strftime('%Y-%m-%d')}-{uuid.uuid4().hex[:8]}",
        "opened_at":          datetime.now(timezone.utc).isoformat(),
        "workforce_id":       score_record["workforce_id"],
        "patient_id":         score_record["patient_id"],
        "tier":               score_record["tier"],
        "composite_score":    float(score_record["composite_calibrated"]),
        "scoring_record_ids":  [score_record["score_id"]],
        "evidence":            evidence,
        "status":              "open_for_review",
        "assigned_to":         None,
        "outcome":             None,
        "outcome_notes":       None,
    }

    table = dynamodb.Table(CASE_STATE_TABLE)
    table.put_item(Item=_decimalize(case))

    eventbridge.put_events(Entries=[{
        "Source":      "access-anomaly.case-builder",
        "DetailType":  "CaseOpened",
        "Detail":      json.dumps({
            "case_id":          case["case_id"],
            "tier":             case["tier"],
            "composite_score":  case["composite_score"],
        }, default=str),
        "EventBusName": CASE_BUS,
    }])

    _emit_metric(f"CasesOpened_{case['tier']}", 1)
    return case["case_id"]

VALID_OUTCOMES = {
    "confirmed_violation",
    "dismissed_legitimate",
    "dismissed_inconclusive",
    "escalated_to_hr",
    "escalated_to_legal",
    "escalated_to_law_enforcement",
}

def initiate_breach_review(case):
    """Start the HIPAA breach-notification workflow.

    Production publishes to an EventBridge rule that triggers a Step
    Functions workflow with breach review, OCR notification clock
    tracking, and legal-coordination handoff. The teaching example
    just emits the event.
    """
    eventbridge.put_events(Entries=[{
        "Source":      "access-anomaly.outcome-capture",
        "DetailType":  "BreachReviewInitiated",
        "Detail":      json.dumps({
            "case_id":      case["case_id"],
            "workforce_id": case["workforce_id"],
            "patient_id":   case["patient_id"],
            "started_at":   datetime.now(timezone.utc).isoformat(),
        }, default=str),
        "EventBusName": CASE_BUS,
    }])

def refer_to_hr(case):
    """Hand off a confirmed-violation case to HR for employment review.

    Production publishes to a separate EventBridge rule that the HR
    integration consumes. The case package (evidence, narrative,
    investigator notes) goes with the referral. Workforce-equity and
    due-process considerations apply.
    """
    eventbridge.put_events(Entries=[{
        "Source":      "access-anomaly.outcome-capture",
        "DetailType":  "HRReferral",
        "Detail":      json.dumps({
            "case_id":      case["case_id"],
            "workforce_id": case["workforce_id"],
            "tier":         case["tier"],
            "referred_at":  datetime.now(timezone.utc).isoformat(),
        }, default=str),
        "EventBusName": CASE_BUS,
    }])

def update_user_state_with_prior_violation(workforce_id):
    """Mark the user as having a prior confirmed violation.

    Used as a feature in subsequent scoring (a user with a recent
    confirmed violation gets elevated scrutiny within policy bounds).
    Subject to organizational policy; some programs do not retain this
    flag indefinitely.
    """
    table = dynamodb.Table(USER_STATE_TABLE)
    table.update_item(
        Key={"workforce_id": workforce_id},
        UpdateExpression="SET prior_violation = :v, prior_violation_at = :t",
        ExpressionAttributeValues={
            ":v": True,
            ":t": datetime.now(timezone.utc).isoformat(),
        },
    )

def on_investigator_action(action):
    """Record an investigator's adjudication and feed the learning loop.

    action keys:
      case_id, outcome, notes, investigator_id
      dismissal_reason (when outcome is dismissed_legitimate)
    """
    if action["outcome"] not in VALID_OUTCOMES:
        raise ValueError(f"invalid outcome: {action['outcome']}")

    table = dynamodb.Table(CASE_STATE_TABLE)
    response = table.get_item(Key={"case_id": action["case_id"]})
    case = _undecimalize(response.get("Item"))
    if case is None:
        logger.warning("outcome for unknown case",
                       extra={"case_id": action["case_id"]})
        return None

    case["outcome"]      = action["outcome"]
    case["outcome_notes"] = action.get("notes")
    case["outcome_at"]   = datetime.now(timezone.utc).isoformat()
    case["assigned_to"]  = action["investigator_id"]
    case["status"]        = "closed"

    if action["outcome"] == "confirmed_violation":
        initiate_breach_review(case)
        refer_to_hr(case)
        update_user_state_with_prior_violation(case["workforce_id"])

    if action["outcome"] == "dismissed_legitimate":
        add_suppression_rule(
            workforce_id=case["workforce_id"],
            patient_id=case["patient_id"],
            reason=action.get("dismissal_reason", "investigator_cleared"),
            valid_for_days=SUPPRESSION_AFTER_DISMISSAL_DAYS,
        )

    table.put_item(Item=_decimalize(case))

    label_record = {
        "label_id":              str(uuid.uuid4()),
        "case_id":               case["case_id"],
        "workforce_id":          case["workforce_id"],
        "patient_id":            case["patient_id"],
        "scoring_record_ids":    case.get("scoring_record_ids", []),
        "composite_score":       float(case["composite_score"]),
        "tier":                  case["tier"],
        "outcome":               case["outcome"],
        "outcome_at":            case["outcome_at"],
        "time_to_adjudication_seconds": (
            (datetime.fromisoformat(case["outcome_at"].replace("Z", "+00:00"))
             - datetime.fromisoformat(case["opened_at"].replace("Z", "+00:00"))).total_seconds()
        ),
        "label":                 1 if case["outcome"] == "confirmed_violation" else 0,
        "investigator_id":       case["assigned_to"],
    }

    s3_client.put_object(
        Bucket=TRAINING_LABELS_BUCKET,
        Key=(
            f"outcomes/year={case['outcome_at'][:4]}/"
            f"month={case['outcome_at'][5:7]}/"
            f"{label_record['label_id']}.json"
        ),
        Body=json.dumps(label_record, default=str).encode("utf-8"),
        ServerSideEncryption="aws:kms",
    )

    eventbridge.put_events(Entries=[{
        "Source":      "access-anomaly.outcome-capture",
        "DetailType":  "CaseClosed",
        "Detail":      json.dumps(label_record, default=str),
        "EventBusName": CASE_BUS,
    }])

    _emit_metric(f"Outcome_{case['outcome']}", 1)
    return label_record