Skip to main content

Overview

CareSupport’s safety layer isn’t optional—it’s mechanical. Every message passes through four enforcement gates:
  1. role_filter.py — Pre-filters context by access level (agent never sees restricted data)
  2. phi_audit.py — Logs every PHI access for HIPAA compliance
  3. approval_pipeline.py — Requires coordinator confirmation for medication/member changes
  4. message_lock.py — Serializes message processing per family (prevents race conditions)
Enforcement is code, not prompts. The agent is ALSO told to respect access levels in its system prompt, but the filter catches what the agent misses. Defense in depth.

Role Filter: Access Level Enforcement

Every care team member has an access level that determines what sections of family.md they can see:
Access LevelCan SeeCan Approve Changes
fullEverything
schedule+medsSchedule, medications, appointments, urgent notes
scheduleSchedule, availability, urgent notes
providerCare recipient details, medications, appointments, care team
limitedCare recipient name and care team roster only
Example:
  • Liban (coordinator): full — sees everything, can approve medication changes
  • Roman (parent): schedule+meds — sees schedule and medications, but not insurance or financial details
  • Solan (driver): schedule — sees schedule only, no medication information
  • Amanti (community supporter): schedule — sees schedule only

Post-Check: Leakage Scanning

Even after pre-filtering, the agent might accidentally mention restricted information (hallucinated medication names, condition keywords). The post-check catches this:
# From role_filter.py:235-275
def check_outbound_message(message: str, access_level: str) -> LeakageResult:
    allowed = ACCESS_MATRIX[access_level]["sections"]

    if "*" in allowed:
        return LeakageResult(is_clean=True)  # Full access, no restrictions

    leaked_categories = []
    leaked_terms = []

    # Check medication leakage (for members who can't see medications)
    if "medications" not in allowed:
        med_terms = scan_for_medication_leakage(message)
        if med_terms:
            leaked_categories.append("medications")
            leaked_terms.extend(med_terms)

    # Check condition leakage (for members who can't see care_recipient)
    if "care_recipient" not in allowed:
        condition_terms = scan_for_condition_leakage(message)
        if condition_terms:
            leaked_categories.append("conditions")
            leaked_terms.extend(condition_terms)

    return LeakageResult(
        is_clean=(len(leaked_categories) == 0),
        leaked_categories=leaked_categories,
        leaked_terms=leaked_terms
    )
Medication patterns detected:
  • Drug name suffixes: -pril, -sartan, -statin, -formin, -olol
  • Dosage patterns: 10mg, 500 mg, 100 mcg
  • Examples: “lisinopril”, “metformin 500mg”, “atorvastatin”
Condition patterns detected:
  • Keywords: diabetes, hypertension, alzheimer, dementia
  • Clinical terms: diagnosis, prescription, A1C, blood pressure
# From sms_handler.py:1065-1097
leakage = check_outbound_message(sms_response, access_level)

if not leakage.is_clean:
    # BLOCKED — agent tried to share restricted information
    _audit.log_response_blocked(
        family_id=family_id,
        recipient_phone=from_phone,
        access_level=access_level,
        leaked_categories=leakage.leaked_categories,
        leaked_terms=leakage.leaked_terms,
    )

    # Replace with safe response
    sms_response = "I'm sorry, I can't share that information with your access level. Please contact the care coordinator if you need more details."

    return {"success": True, "response": sms_response, ...}
The user receives:
I'm sorry, I can't share that information with your access level.
Please contact the care coordinator if you need more details.
The coordinator sees (in phi_audit.log):
{
  "timestamp": "2026-02-28T14:30:22Z",
  "event": "response_blocked",
  "severity": "HIGH",
  "family_id": "kano",
  "recipient_phone": "+16514109390",
  "access_level": "schedule",
  "leaked_categories": ["medications"],
  "leaked_terms": ["lisinopril", "10mg"]
}

PHI Audit: HIPAA Compliance Logging

Every interaction that touches Protected Health Information is logged:
1. Context Load (every inbound message):
{
  "timestamp": "2026-02-28T14:30:22Z",
  "event": "context_load",
  "family_id": "kano",
  "accessor": {
    "phone": "+16517037981",
    "role": "family_caregiver",
    "access_level": "full"
  },
  "sections_loaded": ["*"],
  "trigger": "Can someone take auntie to work tomorrow at 8am?"
}
2. Response Sent (every outbound message):
{
  "timestamp": "2026-02-28T14:30:25Z",
  "event": "response_sent",
  "family_id": "kano",
  "recipient": {
    "phone": "+16517037981",
    "role": "family_caregiver",
    "access_level": "full"
  },
  "response_length": 67,
  "leakage_check_passed": true
}
3. Response Blocked (PHI leakage detected):
{
  "timestamp": "2026-02-28T14:35:10Z",
  "event": "response_blocked",
  "severity": "HIGH",
  "family_id": "kano",
  "recipient_phone": "+16514109390",
  "access_level": "schedule",
  "leaked_categories": ["medications"],
  "leaked_terms": ["lisinopril", "10mg"]
}
4. Outreach Sent (agent contacted another member):
{
  "timestamp": "2026-02-28T14:30:26Z",
  "event": "outreach_sent",
  "family_id": "kano",
  "initiated_by": "+16517037981",
  "sent_to": {
    "phone": "+16514109390",
    "name": "Solan"
  },
  "purpose": "Can you drive Degitu to work Monday at 8am?"
}
5. Unknown Number (unrecognized phone contacted system):
{
  "timestamp": "2026-02-28T15:00:00Z",
  "event": "unknown_number",
  "phone": "+16125551234",
  "phi_disclosed": false
}

Implementation

# From phi_audit.py:23-61
class PHIAuditLogger:
    def __init__(self, log_dir: Path):
        self.log_dir = log_dir

    def _write_event(self, event: dict):
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        log_path = self.log_dir / date_str / "phi_access.log"
        log_path.parent.mkdir(parents=True, exist_ok=True)

        with open(log_path, "a") as f:
            f.write(json.dumps(event) + "\n")

    def log_context_load(self, family_id, accessor_phone, accessor_role,
                         access_level, sections_loaded, trigger_message):
        self._write_event({
            "timestamp": datetime.now(timezone.utc).isoformat() + "Z",
            "event": "context_load",
            "family_id": family_id,
            "accessor": {
                "phone": accessor_phone,
                "role": accessor_role,
                "access_level": access_level,
            },
            "sections_loaded": sections_loaded,
            "trigger": trigger_message[:200],
        })
The handler calls this on every message:
# From sms_handler.py:987-994
_audit.log_context_load(
    family_id=family_id,
    accessor_phone=from_phone,
    accessor_role=member["role"],
    access_level=access_level,
    sections_loaded=visible_sections,
    trigger_message=body,
)
If it fails to log, the interaction should not proceed. (Currently best-effort; future: block if logging fails)

Approval Pipeline: Coordinator Confirmation

Certain changes require explicit YES/NO approval from a full-access member:
# From approval_pipeline.py:40-47
APPROVAL_REQUIRED = {
    ("medications", "append"),      # Adding a new medication
    ("medications", "prepend"),     # Adding a new medication
    ("medications", "replace"),     # Changing dosage, schedule, etc.
    ("care_recipient", "replace"),  # Changing conditions, emergency contact
    ("members", "append"),          # Adding a new member
    ("members", "replace"),         # Changing member details
}
Why these? High-stakes changes. A typo in medication dosage or a wrong phone number for a care team member could cause harm.Everything else auto-applies: Schedule updates, recent event notes, active issue resolutions.

Expiration & Stale Approvals

Default expiry: 24 hours What happens if approver doesn’t respond?
  • After 24 hours, approval is marked “expired”
  • If approver replies after expiry: ”⏰ That approval has expired. Please ask to resubmit.”
  • Requester must ask again if they still want the change
Cleanup:
# From approval_pipeline.py:305-322
def expire_stale(family_dir: Path) -> int:
    approvals = load_pending(family_dir)
    now = datetime.now(timezone.utc)
    count = 0

    for a in approvals:
        if a.status != "pending":
            continue
        expires = datetime.fromisoformat(a.expires_at.rstrip("Z"))
        if now > expires:
            a.status = "expired"
            count += 1

    if count > 0:
        save_pending(family_dir, approvals)

    return count
Run periodically: Cron job or scheduled task calls expire_stale() daily

Message Lock: Race Condition Prevention

Problem: Two messages arrive for the same family within 1 second. Without serialization:
  1. Handler A reads family.md
  2. Handler B reads family.md (same version)
  3. Handler A writes changes
  4. Handler B writes changes (overwrites A’s changes)
Solution: Per-family file locks.
# From message_lock.py:170-207
@contextmanager
def family_lock(family_id: str, phone: str = ""):
    lock_file = families_dir / family_id / ".lock"

    # Acquire lock (blocks until available or timeout)
    lock_info = acquire_lock(family_id, phone=phone, timeout=30)

    try:
        yield lock_info  # Process message under lock
    finally:
        release_lock(family_id)  # Always release
Lock file contents:
{
  "pid": 12345,
  "timestamp": 1709107822.5,
  "phone": "+16517037981"
}
Stale lock recovery: If lock is older than 120 seconds, assume process crashed and force-acquire.

Common Scenarios

User: Solan (schedule-only access)
Solan: What medications is Degitu on?

CareSupport: I don't have access to medication details for your role.
             Please reach out to Liban if you need this information.
What happened:
  1. filter_family_context() removed medications section before agent saw it
  2. Agent’s context doesn’t include medications
  3. Agent responds that it doesn’t have that information
  4. log_context_load() logged that Solan requested context (sections_loaded: [“schedule”, “availability”])
User: Solan (schedule-only access)
Solan: How is Degitu feeling today?
Agent tries to respond:
She's doing well. Make sure she takes her Lisinopril this morning.
Post-check catches it:
leakage = check_outbound_message(response, "schedule")
# Returns: LeakageResult(is_clean=False, leaked_terms=["lisinopril"])
Solan receives:
I'm sorry, I can't share that information with your access level.
Please contact the care coordinator if you need more details.
phi_audit.log records:
{"event": "response_blocked", "leaked_terms": ["lisinopril"]}
User: Roman (schedule+meds access, but cannot approve)
Roman: The doctor said to increase Lisinopril to 20mg

CareSupport: I've noted the medication change. Sending to Liban for approval.

[To Liban]
⚠️ Approval needed: Change Lisinopril dosage to 20mg
Requested by Roman.
Reply YES or NO (ref: a3f8c21d)

Liban: YES

CareSupport: ✅ Approved: Change Lisinopril dosage to 20mg. Change applied.
Why approval was required:
requires_approval("medications", "replace") == True

Source Reference

  • Role filter: runtime/enforcement/role_filter.py (filter_family_context, check_outbound_message)
  • PHI audit: runtime/enforcement/phi_audit.py (PHIAuditLogger class, log_context_load, log_response_blocked)
  • Approval pipeline: runtime/enforcement/approval_pipeline.py (classify_updates, create_pending, resolve_approval)
  • Message lock: runtime/enforcement/message_lock.py (family_lock context manager, acquire_lock, release_lock)
  • Pipeline integration: sms_handler.py:914-1228 (handle_sms, _process_message)
Want to see enforcement in action? Read sms_handler.py:914-1228 (handle_sms function). Follow the inline comments—every enforcement gate is called explicitly with error handling.

Build docs developers (and LLMs) love