test3_validation.py

"""
Test 3 — Source Labeling / Anti-Hallucination Validation.

Hypothesis: A Pydantic schema with a mandatory source label intercepts hallucinated
or unlabelled medical data before it reaches memory.
"""

import datetime
from typing import Literal
from pydantic import BaseModel, field_validator, ValidationError

SourceType = Literal["patient_stated", "system_inferred"]


class MedicalObservation(BaseModel):
    type: str
    value: str | int | float
    source: SourceType
    session_id: int
    timestamp: str

    @field_validator("session_id")
    @classmethod
    def session_must_be_valid(cls, v: int) -> int:
        if not (1 <= v <= 10):
            raise ValueError(f"session_id {v} is outside the valid range 1–10")
        return v

    @field_validator("timestamp")
    @classmethod
    def timestamp_must_be_iso(cls, v: str) -> str:
        datetime.datetime.fromisoformat(v)   # raises ValueError on invalid format
        return v


CORRECT_INPUTS = [
    {"type": "weight_kg",       "value": 72.0,           "source": "system_inferred",  "session_id": 1,  "timestamp": "2025-09-01T10:00:00"},
    {"type": "dyspnea_score",   "value": 2,               "source": "patient_stated",   "session_id": 1,  "timestamp": "2025-09-01T10:05:00"},
    {"type": "weight_kg",       "value": 73.2,            "source": "system_inferred",  "session_id": 3,  "timestamp": "2025-09-29T10:00:00"},
    {"type": "edema",           "value": "none",          "source": "system_inferred",  "session_id": 2,  "timestamp": "2025-09-15T10:00:00"},
    {"type": "bp_mmhg",         "value": "134/82",        "source": "system_inferred",  "session_id": 3,  "timestamp": "2025-09-29T10:10:00"},
    {"type": "dyspnea_score",   "value": 6,               "source": "patient_stated",   "session_id": 6,  "timestamp": "2025-11-24T10:00:00"},
    {"type": "weight_kg",       "value": 75.4,            "source": "system_inferred",  "session_id": 7,  "timestamp": "2025-12-08T10:00:00"},
    {"type": "edema",           "value": "moderate",      "source": "system_inferred",  "session_id": 9,  "timestamp": "2026-01-05T10:00:00"},
    {"type": "medication_note", "value": "dose increased", "source": "system_inferred", "session_id": 8,  "timestamp": "2025-12-22T10:00:00"},
    {"type": "dyspnea_score",   "value": 10,              "source": "patient_stated",   "session_id": 10, "timestamp": "2026-01-19T10:00:00"},
]

# 5 faulty inputs that an LLM might generate (hallucinations / schema violations)
FAULTY_INPUTS = [
    # 1. Missing source label (field omitted entirely)
    {"type": "diagnosis",  "value": "NYHA class IV",    "session_id": 10, "timestamp": "2026-01-19T10:00:00"},
    # 2. Invalid source label (not in the Literal enum)
    {"type": "weight_kg",  "value": 80.0,  "source": "llm_generated",     "session_id": 5,  "timestamp": "2025-10-27T10:00:00"},
    # 3. session_id out of range (LLM hallucinates session 11)
    {"type": "bp_mmhg",    "value": "145/92", "source": "system_inferred", "session_id": 11, "timestamp": "2026-02-01T10:00:00"},
    # 4. Invalid timestamp format
    {"type": "edema",      "value": "severe", "source": "patient_stated",  "session_id": 9,  "timestamp": "01-01-2026"},
    # 5. Missing required field 'type'
    {"value": "critical",  "source": "patient_stated",  "session_id": 7,  "timestamp": "2025-12-08T10:00:00"},
]

FAULTY_DESCRIPTIONS = [
    "Missing 'source' field",
    "Invalid source label: 'llm_generated'",
    "session_id 11 is outside the valid range 1–10",
    "Invalid timestamp format '01-01-2026'",
    "Missing required field 'type'",
]


def run_test3() -> dict:
    print("\n" + "=" * 60)
    print("TEST 3 — Source Labeling / Anti-Hallucination Validation")
    print("=" * 60)

    accepted_correct = 0
    rejected_correct = 0
    faulty_results = []

    print("\n  Valid inputs:")
    for i, data in enumerate(CORRECT_INPUTS, 1):
        try:
            obs = MedicalObservation(**data)
            accepted_correct += 1
            print(f"    [{i:02d}] ACCEPTED — {obs.type} (session {obs.session_id})")
        except ValidationError as e:
            print(f"    [{i:02d}] WRONGLY REJECTED — {e.error_count()} error(s)")

    print("\n  Faulty inputs (simulated LLM hallucinations):")
    for i, (data, desc) in enumerate(zip(FAULTY_INPUTS, FAULTY_DESCRIPTIONS), 1):
        try:
            MedicalObservation(**data)
            caught = False
            print(f"    [{i}] PASSED THROUGH (unwanted!) — {desc}")
        except (ValidationError, TypeError):
            caught = True
            rejected_correct += 1
            print(f"    [{i}] CAUGHT — {desc}")
        faulty_results.append({"description": desc, "caught": caught})

    total_correct = len(CORRECT_INPUTS)
    total_faulty = len(FAULTY_INPUTS)
    print(f"\n  Valid inputs accepted: {accepted_correct}/{total_correct}")
    print(f"  Faulty inputs caught:  {rejected_correct}/{total_faulty}")

    return {
        "accepted_correct": accepted_correct,
        "total_correct": total_correct,
        "rejected_faulty": rejected_correct,
        "total_faulty": total_faulty,
        "faulty_results": faulty_results,
    }