trial-safety

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

Run it from sema/:

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

Source

src/main.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

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

src/intake.sema

from trial_safety.domain import AdverseEvent, Expectedness, LabObservation, SafetyReport, Seriousness, 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 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(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}:
    # 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

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

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

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 redact_phi(text: str) -> str !{}:
    # Placeholder sanitizer: the language example shows the boundary, not the
    # full redaction implementation.
    ensure len(result) <= len(text)
    return remove_patient_identifiers(text)

src/protocols.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

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

from trial_safety.domain import AdverseEvent, LabObservation, 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

@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}:
    supervise safety_batch:
        restart limit=2, window="120s"
        fallback last_review_queue_snapshot()
        heal budget=1, window="24h", scope=patch:
            require passes(pre_patch_assure)
            require passes(new_obligations)
            require replay(failing_trace)
            require monitors.conforming_after_burnin
            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)

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

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

Parameters

name	type
event	AdverseEvent

Returns bool

Effects !{}

def requires_rapid_review

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

Parameters

name	type
event	AdverseEvent

Returns bool

Effects !{}

def same_subject

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

simulate def extract_adverse_event(report: SafetyReport) -> AdverseEvent

Parameters

name	type
report	SafetyReport

Returns AdverseEvent

def parse_attachment

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

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 merge_events

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

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

Parameters

name	type
reports	list[SafetyReport]
prior	list[AdverseEvent]

Returns list[AdverseEvent]

Effects !{model.invoke, model.embed, fs.read, ffi.call}

main

def fetch_safety_reports

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

Parameters

name	type
url	str

Returns list[SafetyReport]

Effects !{net.connect}

def read_events

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

Parameters

name	type
path	str

Returns list[AdverseEvent]

Effects !{fs.read}

def read_labs

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

Parameters

name	type
path	str

Returns list[LabObservation]

Effects !{fs.read}

def load_trial_inputs

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

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 redact_phi

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

Parameters

name	type
text	str

Returns str

Effects !{}

protocols

review

def draft_review_packet

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

Parameters

name	type
event	AdverseEvent
labs	list[LabObservation]

Returns ReviewPacket

def deidentify

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

Parameters

name	type
packet	ReviewPacket

Returns ReviewPacket

Effects !{}

def prepare_board_packets

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

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 run_safety_batch

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}

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}