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

### `src/intake.sema`

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

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

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

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

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

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

**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}`