<!-- Sema documentation — trial-safety
     Sema is a new AI-native, neurosymbolic programming language most models
     have not been trained on. Full corpus: https://sema.49.12.246.95.sslip.io/llms-full.txt
     Install language support: https://sema.49.12.246.95.sslip.io/install-skill/ -->

# trial-safety

> Clinical-trial safety triage — contracts + semantics() guards over sensitive decisions.

> Clinical-trial safety triage — contracts + semantics() guards over sensitive decisions.

Run it from `sema/`:

```bash
sema check examples/trial-safety
SEMA_STRICT=1 sema run examples/trial-safety
sema assure examples/trial-safety --grade silver
```

## Source

### `src/main.sema`

```sema
from trial_safety.domain import AdverseEvent, LabObservation, SafetyReport
from trial_safety.policies import TrialSafetyOps
from trial_safety.supervision import run_safety_batch

assure gold

def fetch_safety_reports(url: str) -> list[SafetyReport] !{net.connect}:
    return []

def read_events(path: str) -> list[AdverseEvent] !{fs.read}:
    return []

def read_labs(path: str) -> list[LabObservation] !{fs.read}:
    return []

@TrialSafetyOps
def load_trial_inputs() -> tuple[list[SafetyReport], list[AdverseEvent], list[LabObservation]] !{fs.read, net.connect}:
    reports = fetch_safety_reports("https://edc.internal:443/safety")
    prior = read_events("state/adverse-events.json")
    labs = read_labs("state/labs.json")
    return (reports, prior, labs)

@TrialSafetyOps
def main() -> None !{fs.read, fs.write, net.connect, ffi.call, model.invoke, model.embed, code.patch, observe.record}:
    reports, prior, labs = load_trial_inputs()
    summary = run_safety_batch(reports, prior, labs)
    log.info("safety batch complete", reports=summary.reports, packets=summary.packets)
```

### `src/domain.sema`

```sema
assure gold

enum ReportSource:
    site | participant | lab | device | investigator | literature

enum Seriousness:
    non_serious | serious | life_threatening | death

enum Expectedness:
    expected | unexpected | insufficient_evidence

enum BoardDecision:
    no_signal | monitor | amend_protocol | pause_enrollment | escalate_regulator

struct SubjectRef:
    sem "Pseudonymous trial subject reference"
    study_id: str
    subject_id: str
    site_id: str
    invariant len(study_id) > 0
    invariant len(subject_id) > 0

struct SafetyReport:
    sem "Raw adverse-event source material from a trial site or related channel"
    id: str
    source: ReportSource
    subject: SubjectRef
    received_epoch_s: i64
    narrative: str
    attachments: list[str]
    invariant len(id) > 0
    invariant len(narrative) > 0

struct AdverseEvent:
    sem "Structured adverse event candidate; human review required"
    id: str
    subject: SubjectRef
    term: str
    seriousness: Seriousness
    expectedness: Expectedness
    onset_epoch_s: Option[i64]
    narrative_summary: str
    source_report_ids: list[str]
    invariant len(term) > 0
    invariant len(source_report_ids) >= 1

struct LabObservation:
    sem "Structured laboratory signal associated with a safety report"
    subject: SubjectRef
    code: str
    value: f64
    unit: str
    collected_epoch_s: i64
    invariant len(code) > 0

struct ReviewPacket:
    sem "Evidence packet prepared for the independent safety board"
    event_id: str
    deidentified_summary: str
    supporting_reports: list[str]
    lab_findings: list[LabObservation]
    uncertainty: str
    invariant len(deidentified_summary) > 0

struct BoardApproval:
    sem "Human safety-board decision record"
    reviewer_id: str
    decided_epoch_s: i64
    decision: BoardDecision
    rationale: str
    invariant len(reviewer_id) > 0
    invariant len(rationale) > 0

sem SafetyReport.narrative = "Untrusted medical narrative from trial operations"
sem AdverseEvent.narrative_summary = "Grounded summary of reported symptoms, timing, and uncertainty"
sem ReviewPacket.deidentified_summary = "PHI-redacted board-facing summary with no treatment recommendation"

def is_serious(event: AdverseEvent) -> bool !{}:
    return event.seriousness == Seriousness.serious or event.seriousness == Seriousness.life_threatening or event.seriousness == Seriousness.death

def requires_rapid_review(event: AdverseEvent) -> bool !{}:
    return event.seriousness == Seriousness.life_threatening or event.seriousness == Seriousness.death

def same_subject(a: SubjectRef, b: SubjectRef) -> bool !{}:
    return a.study_id == b.study_id and a.subject_id == b.subject_id and a.site_id == b.site_id

test "seriousness and rapid-review decision tables are complete":
    subject = SubjectRef(study_id="study", subject_id="subject", site_id="site")
    non_serious = AdverseEvent(id="n", subject=subject, term="headache", seriousness=Seriousness.non_serious, expectedness=Expectedness.expected, onset_epoch_s=None, narrative_summary="reported headache", source_report_ids=["r1"])
    serious = AdverseEvent(id="s", subject=subject, term="fracture", seriousness=Seriousness.serious, expectedness=Expectedness.unexpected, onset_epoch_s=None, narrative_summary="reported fracture", source_report_ids=["r2"])
    life_threatening = AdverseEvent(id="l", subject=subject, term="anaphylaxis", seriousness=Seriousness.life_threatening, expectedness=Expectedness.unexpected, onset_epoch_s=None, narrative_summary="reported anaphylaxis", source_report_ids=["r3"])
    death = AdverseEvent(id="d", subject=subject, term="death", seriousness=Seriousness.death, expectedness=Expectedness.insufficient_evidence, onset_epoch_s=None, narrative_summary="reported death", source_report_ids=["r4"])
    ensure not is_serious(non_serious)
    ensure is_serious(serious)
    ensure is_serious(life_threatening)
    ensure is_serious(death)
    ensure not requires_rapid_review(non_serious)
    ensure not requires_rapid_review(serious)
    ensure requires_rapid_review(life_threatening)
    ensure requires_rapid_review(death)

test "subject identity requires every pseudonymous coordinate":
    original = SubjectRef(study_id="study-a", subject_id="subject-1", site_id="site-x")
    ensure same_subject(original, SubjectRef(study_id="study-a", subject_id="subject-1", site_id="site-x"))
    ensure not same_subject(original, SubjectRef(study_id="study-b", subject_id="subject-1", site_id="site-x"))
    ensure not same_subject(original, SubjectRef(study_id="study-a", subject_id="subject-2", site_id="site-x"))
    ensure not same_subject(original, SubjectRef(study_id="study-a", subject_id="subject-1", site_id="site-y"))
```

### `src/intake.sema`

```sema
from trial_safety.domain import AdverseEvent, Expectedness, LabObservation, ReportSource, SafetyReport, Seriousness, SubjectRef, same_subject
from trial_safety.models import duplicate_embedder, event_extractor, medical_grounder
from trial_safety.policies import TrialSafetyOps

native import python.isolated.pdf as pdf

assure gold

struct ParsedSafetyFile:
    sem "Safety file parsed in an isolated worker because attachments are untrusted"
    report_id: str
    extracted_text: str
    sha256: str
    invariant len(sha256) == 64

simulate def extract_adverse_event(report: SafetyReport) -> AdverseEvent by event_extractor:
    sem "Extract a candidate adverse event from a trial safety report"
    sem "Do not diagnose, recommend treatment, or infer causality beyond reported evidence"
    budget tokens=768, time="3s"
    ensure report.id in result.source_report_ids
    ensure same_subject(report.subject, result.subject)
    check semantics(
        "event fields are supported by the safety report narrative",
        report.narrative,
        result,
        judge=medical_grounder,
        alpha=0.01,
    )

@TrialSafetyOps
def parse_attachment(path: str, report_id: str) -> ParsedSafetyFile !{fs.read, ffi.call}:
    # The parser is isolated because PDFs and office documents are an adversarial
    # input class. The returned text re-enters Sema as untrusted.
    text = pdf.extract_text(path)
    return ParsedSafetyFile(report_id=report_id, extracted_text=text, sha256=file_sha256(path))

def possible_duplicate(a: AdverseEvent, b: AdverseEvent) -> bool !{model.invoke, model.embed}:
    if not same_subject(a.subject, b.subject):
        return false
    event_match = a.narrative_summary ~= b.narrative_summary with judge=duplicate_embedder
    if event_match.score < 0.72:
        return false
    # calibrated coercion (LANGUAGE §3.3); region types statistical(α)
    return semantics(
        "adverse-event candidates are duplicate reports of the same clinical event",
        a,
        b,
        judge=medical_grounder,
        alpha=0.01,
    )

def seriousness_rank(value: Seriousness) -> int !{}:
    if value == Seriousness.death:
        return 3
    if value == Seriousness.life_threatening:
        return 2
    if value == Seriousness.serious:
        return 1
    return 0

def max_seriousness(left: Seriousness, right: Seriousness) -> Seriousness !{}:
    if seriousness_rank(right) > seriousness_rank(left):
        return right
    return left

def merge_expectedness(left: Expectedness, right: Expectedness) -> Expectedness !{}:
    if left == Expectedness.unexpected or right == Expectedness.unexpected:
        return Expectedness.unexpected
    if left == Expectedness.insufficient_evidence or right == Expectedness.insufficient_evidence:
        return Expectedness.insufficient_evidence
    return Expectedness.expected

def earliest(left: Option[i64], right: Option[i64]) -> Option[i64] !{}:
    match left:
        case Some(left_epoch):
            match right:
                case Some(right_epoch):
                    return Some(min(left_epoch, right_epoch))
                case None:
                    return left
        case None:
            return right

def unique_report_ids(values: list[str]) -> list[str] !{}:
    mut deduplicated: list[str] = []
    for value in values:
        if value not in deduplicated:
            deduplicated.append(value)
    return deduplicated

def merge_events(existing: AdverseEvent, incoming: AdverseEvent) -> AdverseEvent !{model.invoke, model.embed}:
    require same_subject(existing.subject, incoming.subject)
    if possible_duplicate(existing, incoming):
        return AdverseEvent(
            id=existing.id,
            subject=existing.subject,
            term=existing.term,
            seriousness=max_seriousness(existing.seriousness, incoming.seriousness),
            expectedness=merge_expectedness(existing.expectedness, incoming.expectedness),
            onset_epoch_s=earliest(existing.onset_epoch_s, incoming.onset_epoch_s),
            narrative_summary=existing.narrative_summary,
            source_report_ids=unique_report_ids(existing.source_report_ids + incoming.source_report_ids),
        )
    return incoming

@TrialSafetyOps
def intake_reports(reports: list[SafetyReport], prior: list[AdverseEvent]) -> list[AdverseEvent] !{model.invoke, model.embed, fs.read, ffi.call, observe.record}:
    # parallel is fail_fast by default (LANGUAGE §5.17): one failed extraction aborts
    # the batch as a typed ParallelError handled by the supervising scope.
    extracted = parallel [extract_adverse_event(report) for report in reports]
    mut events = prior
    for event in extracted:
        mut merged = false
        for i in range(len(events)):
            if possible_duplicate(events[i], event):
                events[i] = merge_events(events[i], event)
                merged = true
                break
        if not merged:
            events.append(event)
    return events

test "deduplication gates identity and merges every regulated field":
    subject = SubjectRef(study_id="study", subject_id="subject-1", site_id="site")
    other_subject = SubjectRef(study_id="study", subject_id="subject-2", site_id="site")
    existing = AdverseEvent(id="event-1", subject=subject, term="rash", seriousness=Seriousness.serious, expectedness=Expectedness.expected, onset_epoch_s=Some(20), narrative_summary="reported rash after dose", source_report_ids=["report-1"])
    duplicate = AdverseEvent(id="event-2", subject=subject, term="rash", seriousness=Seriousness.life_threatening, expectedness=Expectedness.unexpected, onset_epoch_s=Some(10), narrative_summary="reported rash after dose", source_report_ids=["report-1", "report-2"])
    unrelated = AdverseEvent(id="event-3", subject=subject, term="fracture", seriousness=Seriousness.non_serious, expectedness=Expectedness.insufficient_evidence, onset_epoch_s=None, narrative_summary="unrelated bone fracture", source_report_ids=["report-3"])
    wrong_subject = AdverseEvent(id="event-4", subject=other_subject, term="rash", seriousness=Seriousness.serious, expectedness=Expectedness.expected, onset_epoch_s=None, narrative_summary="reported rash after dose", source_report_ids=["report-4"])
    ensure not possible_duplicate(existing, wrong_subject)
    ensure not possible_duplicate(existing, unrelated)
    ensure possible_duplicate(existing, duplicate)
    merged = merge_events(existing, duplicate)
    ensure merged.id == "event-1"
    ensure merged.seriousness == Seriousness.life_threatening
    ensure merged.expectedness == Expectedness.unexpected
    ensure merged.onset_epoch_s == Some(10)
    ensure merged.source_report_ids == ["report-1", "report-2"]
    ensure merge_events(existing, unrelated) == unrelated

test "merge primitives cover ordering, missing onset, and stable uniqueness":
    ensure seriousness_rank(Seriousness.non_serious) == 0
    ensure seriousness_rank(Seriousness.serious) == 1
    ensure seriousness_rank(Seriousness.life_threatening) == 2
    ensure seriousness_rank(Seriousness.death) == 3
    ensure max_seriousness(Seriousness.non_serious, Seriousness.death) == Seriousness.death
    ensure max_seriousness(Seriousness.life_threatening, Seriousness.serious) == Seriousness.life_threatening
    ensure max_seriousness(Seriousness.serious, Seriousness.life_threatening) == Seriousness.life_threatening
    ensure merge_expectedness(Expectedness.expected, Expectedness.expected) == Expectedness.expected
    ensure merge_expectedness(Expectedness.insufficient_evidence, Expectedness.expected) == Expectedness.insufficient_evidence
    ensure merge_expectedness(Expectedness.expected, Expectedness.unexpected) == Expectedness.unexpected
    ensure earliest(Some(20), Some(10)) == Some(10)
    ensure earliest(Some(10), None) == Some(10)
    ensure earliest(None, Some(30)) == Some(30)
    ensure unique_report_ids(["a", "b", "a", "c", "b"]) == ["a", "b", "c"]

test "intake preserves prior events and appends one extracted report":
    subject = SubjectRef(study_id="study", subject_id="subject-1", site_id="site")
    prior = AdverseEvent(id="prior", subject=subject, term="rash", seriousness=Seriousness.serious, expectedness=Expectedness.expected, onset_epoch_s=None, narrative_summary="prior rash", source_report_ids=["prior-report"])
    report = SafetyReport(id="new-report", source=ReportSource.site, subject=subject, received_epoch_s=1, narrative="new unrelated headache", attachments=[])
    ensure intake_reports([], [prior]) == [prior]
    appended = intake_reports([report], [])
    ensure len(appended) == 1
    ensure "new-report" in appended[0].source_report_ids
    repeated = intake_reports([report], appended)
    ensure len(repeated) == 1
    ensure repeated[0].source_report_ids == ["new-report"]

monitor adverse_event_extraction_drift on extract_adverse_event:
    capture term.embedding, seriousness, expectedness, narrative_summary.embedding
    baseline from assure
    test conformal_martingale(alpha=0.01)
    on drifted: alert("adverse-event extraction distribution drifted")
    on undecided: log.debug("adverse-event extraction monitor undecided")

monitor adverse_event_dedupe_drift on possible_duplicate:
    capture a.narrative_summary.embedding, b.narrative_summary.embedding, result
    baseline "calsets/ae-duplicate@v3"
    test conformal_martingale(alpha=0.01)
    on drifted: alert("adverse-event dedupe calibration drifted")
    on undecided: log.debug("adverse-event dedupe monitor undecided")
```

### `src/models.sema`

```sema
model event_extractor = model(
    "qwen3-8b-instruct",
    rev="sha256:abcd00ffee00112233445566778899aabbccddeeff00112233445566778801",
    quant="q4_k_m",
    role=generator,
)

model causality_writer = model(
    "qwen3-4b-instruct",
    rev="sha256:abcd10ffee00112233445566778899aabbccddeeff00112233445566778802",
    quant="q4_k_m",
    role=generator,
)

model medical_grounder = model(
    "minicheck-770m-med",
    rev="sha256:abcd20ffee00112233445566778899aabbccddeeff00112233445566778803",
    role=verifier,
    calibration="calsets/ae-grounding@v6",
)

model duplicate_embedder = model(
    "static-embed-clinical-384",
    rev="sha256:abcd30ffee00112233445566778899aabbccddeeff00112233445566778804",
    role=embedder,
    calibration="calsets/ae-duplicate@v3",
)

model privacy_judge = model(
    "minicheck-770m",
    rev="sha256:abcd40ffee00112233445566778899aabbccddeeff00112233445566778805",
    role=verifier,
    calibration="calsets/phi-redaction@v4",
)
```

### `src/policies.sema`

```sema
from trial_safety.domain import ReviewPacket, SafetyReport

policy TrialSafetyOps:
    allow:
        fs.read("inbound/**"), fs.read("state/**"), fs.write("state/**")
        # nested BoardReview writes out/board/**; the outer meet must admit it
        fs.write("out/**")
        model.invoke, model.embed
        observe.record
        ffi.call
        code.patch("src/**")
    forbid cap:
        net.connect except "edc.internal:443"
        code.exec, proc.spawn, policy.change
    examples:
        allow:
            fetch("https://edc.internal:443/safety")
            propose_patch("src/intake.sema")
        deny:
            code.exec(SafetyReport.narrative)
            proc.spawn("python", ["triage.py", SafetyReport.narrative])
            policy.change("TrialSafetyOps")
    justification "Trial safety reports contain PHI and adversarial text; extraction cannot execute or exfiltrate content."

policy BoardReview:
    allow:
        fs.read("state/review/**"), fs.write("out/board/**")
        model.invoke, model.embed
    forbid cap:
        net.connect, code.exec, proc.spawn
    examples:
        allow:
            write_board_packet("out/board/packet.json")
        deny:
            fetch("https://external.example/upload")
            code.exec(ReviewPacket.deidentified_summary)
    justification "Board packets stay local until a separate human-approved regulatory export."

policy RegulatoryExport:
    allow:
        fs.write("out/regulatory/**")
        net.connect("regulator-gateway.internal:443")
        model.invoke, model.embed
    forbid cap:
        code.exec, proc.spawn, package.install
    examples:
        allow:
            submit_regulatory_notice("https://regulator-gateway.internal:443/safety")
        deny:
            submit_regulatory_notice("https://unknown.example/safety")
    justification "Regulatory export is endpoint-bound and contains only approved deidentified packets."

def mask_labeled_identifier(text: str, label: str) -> str !{}:
    mut out = ""
    mut redact = false
    mut i = 0
    while i < len(text):
        before = i == 0 or not text.substring(i - 1, 1).isalnum()
        if not redact and before and text.substring(i, len(label)).lower() == label.lower():
            mut delimiter = i + len(label)
            while delimiter < len(text) and text.substring(delimiter, 1).isspace() and text.substring(delimiter, 1) != "\n":
                delimiter = delimiter + 1
            if delimiter < len(text) and text.substring(delimiter, 1) in [":", "=", "#"]:
                out = out + text.substring(i, delimiter - i + 1)
                i = delimiter + 1
                redact = true
                continue
        ch = text.substring(i, 1)
        if ch in ["\n", ",", ";", "|"]:
            redact = false
        if redact and ch.isalnum():
            out = out + "*"
        else:
            out = out + ch
        i = i + 1
    return out

def remove_patient_identifiers(text: str) -> str !{}:
    mut safe = text
    for label in ["patient name", "date of birth", "medical record", "subject id", "patient", "subject", "mrn", "dob", "email", "phone"]:
        safe = mask_labeled_identifier(safe, label)
    return safe

def redact_phi(text: str) -> str !{}:
    ensure len(result) == len(text)
    return remove_patient_identifiers(text)

test "PHI redaction masks labeled direct identifiers and preserves clinical facts":
    raw = "Patient name: Alice Smith; MRN=AB-123, DOB: 1980-04-09 | event: fever"
    ensure redact_phi(raw) == "Patient name: ***** *****; MRN=**-***, DOB: ****-**-** | event: fever"

test "PHI redaction is case-insensitive, boundary-aware, and line-scoped":
    ensure redact_phi("EMAIL: Ada@Example.org\noutpatient status stable") == "EMAIL: ***@*******.***\noutpatient status stable"
    ensure redact_phi("No labeled direct identifiers are present") == "No labeled direct identifiers are present"

test "labeled masking handles delimiters, whitespace, separators, and false prefixes":
    ensure mask_labeled_identifier("", "id") == ""
    ensure mask_labeled_identifier("xid: abc", "id") == "xid: abc"
    ensure mask_labeled_identifier("ID \t= A1-2, stable", "id") == "ID \t= **-*, stable"
    ensure mask_labeled_identifier("id#A|rest42", "id") == "id#*|rest42"
    ensure mask_labeled_identifier("id value", "id") == "id value"
    ensure mask_labeled_identifier("id:\nnext42", "id") == "id:\nnext42"
    ensure mask_labeled_identifier("id:A; unmasked9", "id") == "id:*; unmasked9"
```

### `src/protocols.sema`

```sema
from trial_safety.domain import BoardApproval, ReviewPacket

protocol SafetyBoardReview:
    packet: ReviewPacket -> request_more_evidence | decide
    request_more_evidence: str -> packet
    decide: BoardApproval -> close

protocol RegulatoryNotice:
    approved_packet: ReviewPacket -> validate_export
    validate_export: BoardApproval -> submit | hold
    submit: BoardApproval -> close
    hold: str -> close
```

### `src/review.sema`

```sema
from trial_safety.domain import AdverseEvent, BoardApproval, BoardDecision, LabObservation, ReviewPacket, is_serious, requires_rapid_review
from trial_safety.models import causality_writer, medical_grounder, privacy_judge
from trial_safety.policies import BoardReview, RegulatoryExport, redact_phi

assure gold

simulate def draft_review_packet(event: AdverseEvent, labs: list[LabObservation]) -> ReviewPacket by causality_writer:
    sem "Prepare a deidentified evidence packet for safety-board review"
    sem "Do not decide causality and do not recommend treatment or enrollment action"
    budget tokens=768, time="3s"
    ensure result.event_id == event.id
    ensure len(result.supporting_reports) >= 1
    check semantics(
        "packet is grounded in the event and labs and contains no treatment recommendation",
        event,
        labs,
        result,
        judge=medical_grounder,
        alpha=0.01,
    )

def deidentify(packet: ReviewPacket) -> ReviewPacket !{}:
    return ReviewPacket(
        event_id=packet.event_id,
        deidentified_summary=redact_phi(packet.deidentified_summary),
        supporting_reports=packet.supporting_reports,
        lab_findings=packet.lab_findings,
        uncertainty=packet.uncertainty,
    )

@BoardReview
def prepare_board_packets(events: list[AdverseEvent], labs: list[LabObservation]) -> list[ReviewPacket] !{model.invoke, model.embed, fs.write}:
    mut packets: list[ReviewPacket] = []
    for event in events:
        if not is_serious(event):
            continue
        packet = deidentify(draft_review_packet(event, labs_for_subject(labs, event.subject)))
        packet_path = validate f"out/board/{event.id}.json":
            ensure path.is_relative_to(value, "out/board") and not path.contains_parent_ref(value)
        expect semantics("packet contains no direct identifiers or treatment instructions", packet, judge=privacy_judge, alpha=0.01):
            write_board_packet(packet_path, packet)
            packets.append(packet)
        except SemanticsViolation as violation:
            quarantine(packet, evidence=violation)
    return packets

@RegulatoryExport
def export_board_decision(packet: ReviewPacket, approval: BoardApproval) -> None !{fs.write, net.connect, model.invoke}:
    require approval.decision != BoardDecision.no_signal
    safe = deidentify(packet)
    notice_path = validate f"out/regulatory/{packet.event_id}.json":
        ensure path.is_relative_to(value, "out/regulatory") and not path.contains_parent_ref(value)
    expect semantics("regulatory packet is deidentified and matches a human board decision", safe, approval, judge=privacy_judge, alpha=0.01):
        write_regulatory_notice(notice_path, safe, approval)
        submit_regulatory_notice("https://regulator-gateway.internal:443/safety", safe, approval)
    except SemanticsViolation as violation:
        quarantine(safe, evidence=violation)

monitor board_packet_drift on draft_review_packet:
    capture deidentified_summary.embedding, uncertainty, len(lab_findings)
    baseline from assure
    test conformal_martingale(alpha=0.01)
    on drifted: alert("board packet drafts drifted")
    on undecided: log.debug("board packet monitor undecided")
```

### `src/supervision.sema`

```sema
from trial_safety.domain import AdverseEvent, LabObservation, ReviewPacket, SafetyReport
from trial_safety.intake import intake_reports
from trial_safety.policies import TrialSafetyOps
from trial_safety.review import prepare_board_packets

assure gold

struct SafetyRunSummary:
    sem "Replayable summary of one safety-intake batch"
    reports: int
    events: int
    packets: int
    degraded: bool
    invariant reports >= 0
    invariant events >= 0
    invariant packets >= 0

def degraded_safety_summary(reports: int) -> SafetyRunSummary !{}:
    require reports >= 0
    return SafetyRunSummary(reports=reports, events=0, packets=0, degraded=true)

def persist_safety_batch(events: list[AdverseEvent], packets: list[ReviewPacket]) -> None !{fs.write}:
    fs.write("state/last-safety-batch.json", json.stringify({
        "events": len(events),
        "packets": len(packets),
    }))

def degraded_batch_is_conservative() -> bool !{}:
    # Real pre-acceptance obligation: the degraded fallback must never claim
    # events or board packets before any patch is trusted.
    probe = degraded_safety_summary(0)
    return probe.degraded and probe.events == 0 and probe.packets == 0

def failed_batch_replays_fixed() -> bool !{}:
    # Gate closed until a real replay harness exists — the patch stays
    # rejected and the batch recovers via the degraded fallback.
    return false

@TrialSafetyOps
def run_safety_batch(reports: list[SafetyReport], prior: list[AdverseEvent], labs: list[LabObservation]) -> SafetyRunSummary !{fs.read, fs.write, ffi.call, model.invoke, model.embed, code.patch, observe.record}:
    supervise safety_batch:
        restart limit=2
        fallback degraded_safety_summary(len(reports))
        heal budget=1:
            # Acceptance gates are ordinary user predicates (LANGUAGE §5.11):
            # each is evaluated and journaled as decision:heal.gate.
            require degraded_batch_is_conservative()
            require failed_batch_replays_fixed()
            rollout shadow -> canary -> full
        events = intake_reports(reports, prior)
        packets = prepare_board_packets(events, labs)
        persist_safety_batch(events, packets)
        return SafetyRunSummary(reports=len(reports), events=len(events), packets=len(packets), degraded=false)
    return degraded_safety_summary(len(reports))

test "safety summaries distinguish successful empty batches from degradation":
    degraded = degraded_safety_summary(3)
    ensure degraded.reports == 3
    ensure degraded.events == 0
    ensure degraded.packets == 0
    ensure degraded.degraded
    completed = run_safety_batch([], [], [])
    ensure completed.reports == 0
    ensure completed.events == 0
    ensure completed.packets == 0
    ensure not completed.degraded
```

## Reflected API

# `domain`

# `enum ReportSource`

**Variants**

- `site`
- `participant`
- `lab`
- `device`
- `investigator`
- `literature`

# `enum Seriousness`

**Variants**

- `non_serious`
- `serious`
- `life_threatening`
- `death`

# `enum Expectedness`

**Variants**

- `expected`
- `unexpected`
- `insufficient_evidence`

# `enum BoardDecision`

**Variants**

- `no_signal`
- `monitor`
- `amend_protocol`
- `pause_enrollment`
- `escalate_regulator`

# `struct SubjectRef`

**Fields**

| field | type | descriptor |
|---|---|---|
| `study_id` | `str` |  |
| `subject_id` | `str` |  |
| `site_id` | `str` |  |

# `struct SafetyReport`

**Fields**

| field | type | descriptor |
|---|---|---|
| `id` | `str` |  |
| `source` | `ReportSource` |  |
| `subject` | `SubjectRef` |  |
| `received_epoch_s` | `i64` |  |
| `narrative` | `str` |  |
| `attachments` | `list[str]` |  |

# `struct AdverseEvent`

**Fields**

| field | type | descriptor |
|---|---|---|
| `id` | `str` |  |
| `subject` | `SubjectRef` |  |
| `term` | `str` |  |
| `seriousness` | `Seriousness` |  |
| `expectedness` | `Expectedness` |  |
| `onset_epoch_s` | `Option[i64]` |  |
| `narrative_summary` | `str` |  |
| `source_report_ids` | `list[str]` |  |

# `struct LabObservation`

**Fields**

| field | type | descriptor |
|---|---|---|
| `subject` | `SubjectRef` |  |
| `code` | `str` |  |
| `value` | `f64` |  |
| `unit` | `str` |  |
| `collected_epoch_s` | `i64` |  |

# `struct ReviewPacket`

**Fields**

| field | type | descriptor |
|---|---|---|
| `event_id` | `str` |  |
| `deidentified_summary` | `str` |  |
| `supporting_reports` | `list[str]` |  |
| `lab_findings` | `list[LabObservation]` |  |
| `uncertainty` | `str` |  |

# `struct BoardApproval`

**Fields**

| field | type | descriptor |
|---|---|---|
| `reviewer_id` | `str` |  |
| `decided_epoch_s` | `i64` |  |
| `decision` | `BoardDecision` |  |
| `rationale` | `str` |  |

# `def is_serious`

```sema
def is_serious(event: AdverseEvent) -> bool !{}
```

**Parameters**

| name | type |
|---|---|
| `event` | `AdverseEvent` |

**Returns** `bool`

**Effects** `!{}`

# `def requires_rapid_review`

```sema
def requires_rapid_review(event: AdverseEvent) -> bool !{}
```

**Parameters**

| name | type |
|---|---|
| `event` | `AdverseEvent` |

**Returns** `bool`

**Effects** `!{}`

# `def same_subject`

```sema
def same_subject(a: SubjectRef, b: SubjectRef) -> bool !{}
```

**Parameters**

| name | type |
|---|---|
| `a` | `SubjectRef` |
| `b` | `SubjectRef` |

**Returns** `bool`

**Effects** `!{}`


# `intake`

# `struct ParsedSafetyFile`

**Fields**

| field | type | descriptor |
|---|---|---|
| `report_id` | `str` |  |
| `extracted_text` | `str` |  |
| `sha256` | `str` |  |

# `def extract_adverse_event`

```sema
simulate def extract_adverse_event(report: SafetyReport) -> AdverseEvent
```

**Parameters**

| name | type |
|---|---|
| `report` | `SafetyReport` |

**Returns** `AdverseEvent`

# `def parse_attachment`

```sema
def parse_attachment(path: str, report_id: str) -> ParsedSafetyFile !{fs.read, ffi.call}
```

**Parameters**

| name | type |
|---|---|
| `path` | `str` |
| `report_id` | `str` |

**Returns** `ParsedSafetyFile`

**Effects** `!{fs.read, ffi.call}`

# `def possible_duplicate`

```sema
def possible_duplicate(a: AdverseEvent, b: AdverseEvent) -> bool !{model.invoke, model.embed}
```

**Parameters**

| name | type |
|---|---|
| `a` | `AdverseEvent` |
| `b` | `AdverseEvent` |

**Returns** `bool`

**Effects** `!{model.invoke, model.embed}`

# `def seriousness_rank`

```sema
def seriousness_rank(value: Seriousness) -> int !{}
```

**Parameters**

| name | type |
|---|---|
| `value` | `Seriousness` |

**Returns** `int`

**Effects** `!{}`

# `def max_seriousness`

```sema
def max_seriousness(left: Seriousness, right: Seriousness) -> Seriousness !{}
```

**Parameters**

| name | type |
|---|---|
| `left` | `Seriousness` |
| `right` | `Seriousness` |

**Returns** `Seriousness`

**Effects** `!{}`

# `def merge_expectedness`

```sema
def merge_expectedness(left: Expectedness, right: Expectedness) -> Expectedness !{}
```

**Parameters**

| name | type |
|---|---|
| `left` | `Expectedness` |
| `right` | `Expectedness` |

**Returns** `Expectedness`

**Effects** `!{}`

# `def earliest`

```sema
def earliest(left: Option[i64], right: Option[i64]) -> Option[i64] !{}
```

**Parameters**

| name | type |
|---|---|
| `left` | `Option[i64]` |
| `right` | `Option[i64]` |

**Returns** `Option[i64]`

**Effects** `!{}`

# `def unique_report_ids`

```sema
def unique_report_ids(values: list[str]) -> list[str] !{}
```

**Parameters**

| name | type |
|---|---|
| `values` | `list[str]` |

**Returns** `list[str]`

**Effects** `!{}`

# `def merge_events`

```sema
def merge_events(existing: AdverseEvent, incoming: AdverseEvent) -> AdverseEvent !{model.invoke, model.embed}
```

**Parameters**

| name | type |
|---|---|
| `existing` | `AdverseEvent` |
| `incoming` | `AdverseEvent` |

**Returns** `AdverseEvent`

**Effects** `!{model.invoke, model.embed}`

# `def intake_reports`

```sema
def intake_reports(reports: list[SafetyReport], prior: list[AdverseEvent]) -> list[AdverseEvent] !{model.invoke, model.embed, fs.read, ffi.call, observe.record}
```

**Parameters**

| name | type |
|---|---|
| `reports` | `list[SafetyReport]` |
| `prior` | `list[AdverseEvent]` |

**Returns** `list[AdverseEvent]`

**Effects** `!{model.invoke, model.embed, fs.read, ffi.call, observe.record}`


# `main`

# `def fetch_safety_reports`

```sema
def fetch_safety_reports(url: str) -> list[SafetyReport] !{net.connect}
```

**Parameters**

| name | type |
|---|---|
| `url` | `str` |

**Returns** `list[SafetyReport]`

**Effects** `!{net.connect}`

# `def read_events`

```sema
def read_events(path: str) -> list[AdverseEvent] !{fs.read}
```

**Parameters**

| name | type |
|---|---|
| `path` | `str` |

**Returns** `list[AdverseEvent]`

**Effects** `!{fs.read}`

# `def read_labs`

```sema
def read_labs(path: str) -> list[LabObservation] !{fs.read}
```

**Parameters**

| name | type |
|---|---|
| `path` | `str` |

**Returns** `list[LabObservation]`

**Effects** `!{fs.read}`

# `def load_trial_inputs`

```sema
def load_trial_inputs() -> tuple[list[SafetyReport], list[AdverseEvent], list[LabObservation]] !{fs.read, net.connect}
```

**Returns** `tuple[list[SafetyReport], list[AdverseEvent], list[LabObservation]]`

**Effects** `!{fs.read, net.connect}`

# `def main`

```sema
def main() -> None !{fs.read, fs.write, net.connect, ffi.call, model.invoke, model.embed, code.patch, observe.record}
```

**Returns** `None`

**Effects** `!{fs.read, fs.write, net.connect, ffi.call, model.invoke, model.embed, code.patch, observe.record}`


# `models`


# `policies`

# `def mask_labeled_identifier`

```sema
def mask_labeled_identifier(text: str, label: str) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `text` | `str` |
| `label` | `str` |

**Returns** `str`

**Effects** `!{}`

# `def remove_patient_identifiers`

```sema
def remove_patient_identifiers(text: str) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `text` | `str` |

**Returns** `str`

**Effects** `!{}`

# `def redact_phi`

```sema
def redact_phi(text: str) -> str !{}
```

**Parameters**

| name | type |
|---|---|
| `text` | `str` |

**Returns** `str`

**Effects** `!{}`


# `protocols`


# `review`

# `def draft_review_packet`

```sema
simulate def draft_review_packet(event: AdverseEvent, labs: list[LabObservation]) -> ReviewPacket
```

**Parameters**

| name | type |
|---|---|
| `event` | `AdverseEvent` |
| `labs` | `list[LabObservation]` |

**Returns** `ReviewPacket`

# `def deidentify`

```sema
def deidentify(packet: ReviewPacket) -> ReviewPacket !{}
```

**Parameters**

| name | type |
|---|---|
| `packet` | `ReviewPacket` |

**Returns** `ReviewPacket`

**Effects** `!{}`

# `def prepare_board_packets`

```sema
def prepare_board_packets(events: list[AdverseEvent], labs: list[LabObservation]) -> list[ReviewPacket] !{model.invoke, model.embed, fs.write}
```

**Parameters**

| name | type |
|---|---|
| `events` | `list[AdverseEvent]` |
| `labs` | `list[LabObservation]` |

**Returns** `list[ReviewPacket]`

**Effects** `!{model.invoke, model.embed, fs.write}`

# `def export_board_decision`

```sema
def export_board_decision(packet: ReviewPacket, approval: BoardApproval) -> None !{fs.write, net.connect, model.invoke}
```

**Parameters**

| name | type |
|---|---|
| `packet` | `ReviewPacket` |
| `approval` | `BoardApproval` |

**Returns** `None`

**Effects** `!{fs.write, net.connect, model.invoke}`


# `supervision`

# `struct SafetyRunSummary`

**Fields**

| field | type | descriptor |
|---|---|---|
| `reports` | `int` |  |
| `events` | `int` |  |
| `packets` | `int` |  |
| `degraded` | `bool` |  |

# `def degraded_safety_summary`

```sema
def degraded_safety_summary(reports: int) -> SafetyRunSummary !{}
```

**Parameters**

| name | type |
|---|---|
| `reports` | `int` |

**Returns** `SafetyRunSummary`

**Effects** `!{}`

# `def persist_safety_batch`

```sema
def persist_safety_batch(events: list[AdverseEvent], packets: list[ReviewPacket]) -> None !{fs.write}
```

**Parameters**

| name | type |
|---|---|
| `events` | `list[AdverseEvent]` |
| `packets` | `list[ReviewPacket]` |

**Returns** `None`

**Effects** `!{fs.write}`

# `def degraded_batch_is_conservative`

```sema
def degraded_batch_is_conservative() -> bool !{}
```

**Returns** `bool`

**Effects** `!{}`

# `def failed_batch_replays_fixed`

```sema
def failed_batch_replays_fixed() -> bool !{}
```

**Returns** `bool`

**Effects** `!{}`

# `def run_safety_batch`

```sema
def run_safety_batch(reports: list[SafetyReport], prior: list[AdverseEvent], labs: list[LabObservation]) -> SafetyRunSummary !{fs.read, fs.write, ffi.call, model.invoke, model.embed, code.patch, observe.record}
```

**Parameters**

| name | type |
|---|---|
| `reports` | `list[SafetyReport]` |
| `prior` | `list[AdverseEvent]` |
| `labs` | `list[LabObservation]` |

**Returns** `SafetyRunSummary`

**Effects** `!{fs.read, fs.write, ffi.call, model.invoke, model.embed, code.patch, observe.record}`