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

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

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

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 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

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, 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

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 seriousness_rank

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

Parameters

name	type
value	Seriousness

Returns int

Effects !{}

def max_seriousness

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

Parameters

name	type
left	Seriousness
right	Seriousness

Returns Seriousness

Effects !{}

def merge_expectedness

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

Parameters

name	type
left	Expectedness
right	Expectedness

Returns Expectedness

Effects !{}

def earliest

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

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

Parameters

name	type
values	list[str]

Returns list[str]

Effects !{}

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, 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

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 mask_labeled_identifier

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

Parameters

name	type
text	str
label	str

Returns str

Effects !{}

def remove_patient_identifiers

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

Parameters

name	type
text	str

Returns str

Effects !{}

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 degraded_safety_summary

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

Parameters

name	type
reports	int

Returns SafetyRunSummary

Effects !{}

def persist_safety_batch

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

def degraded_batch_is_conservative() -> bool !{}

Returns bool

Effects !{}

def failed_batch_replays_fixed

def failed_batch_replays_fixed() -> bool !{}

Returns bool

Effects !{}

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, 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}