<!-- Sema documentation — robotics-cell
     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/ -->

# robotics-cell

> A robotics work-cell controller with protocols, supervision, and self-healing.

> A robotics work-cell controller with protocols, supervision, and self-healing.

Run it from `sema/`:

```bash
sema check examples/robotics-cell
SEMA_STRICT=1 sema run examples/robotics-cell
sema assure examples/robotics-cell --grade silver
```

## Source

### `src/main.sema`

```sema
from robotics_cell.domain import WorkOrder
from robotics_cell.policies import CellRuntime
from robotics_cell.supervision import run_supervised_order

assure gold

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

@CellRuntime
def main() -> None !{fs.read, fs.write, ffi.call, net.connect, model.invoke, model.embed, code.patch, observe.record}:
    orders = read_work_orders("config/orders.json")
    for order in orders:
        pick = resolve_bin_pose(order.source_bin)
        place = resolve_bin_pose(order.target_bin)
        summary = run_supervised_order(order, pick, place)
        log.info("order complete", order=summary.order_id, faulted=summary.faulted)
```

### `src/domain.sema`

```sema
assure gold

enum CellMode:
    startup | automatic | degraded | manual_hold | emergency_stop

enum RobotState:
    idle | moving | gripping | blocked | faulted | safe_stopped

enum FaultKind:
    slip | collision_risk | unreachable_pose | vision_drift | plc_timeout | unknown

struct Pose:
    sem "Six-degree robot pose in cell coordinates"
    x_mm: f64
    y_mm: f64
    z_mm: f64
    roll_rad: f64
    pitch_rad: f64
    yaw_rad: f64

struct JointVector:
    sem "Joint angles for a six-axis manipulator"
    values_rad: list[f64]
    invariant len(values_rad) == 6

struct TelemetryFrame:
    sem "One timestamped control-loop observation"
    epoch_us: i64
    pose: Pose
    joints: JointVector
    gripper_force_n: f32
    vibration_rms: f32
    state: RobotState
    invariant epoch_us >= 0
    invariant gripper_force_n >= 0.0
    invariant vibration_rms >= 0.0

struct WorkOrder:
    sem "A warehouse movement request accepted by the cell controller"
    id: str
    source_bin: str
    target_bin: str
    sku: str
    max_latency_ms: int
    invariant len(id) > 0
    invariant max_latency_ms > 0

struct MotionSegment:
    sem "Deterministic low-level motion segment"
    start: Pose
    finish: Pose
    max_velocity_mm_s: f32
    max_accel_mm_s2: f32
    invariant max_velocity_mm_s > 0.0
    invariant max_accel_mm_s2 > 0.0

struct MotionPlan:
    sem "Verified plan submitted to the hardware driver"
    order_id: str
    segments: list[MotionSegment]
    expected_duration_ms: int
    safety_margin_mm: f32
    invariant len(segments) >= 1
    invariant expected_duration_ms > 0
    invariant safety_margin_mm >= 0.0

struct FaultEvent:
    sem "Cell fault with enough context for replay and recovery"
    order_id: str
    kind: FaultKind
    observed: TelemetryFrame
    message: str
    invariant len(message) > 0

struct RecoveryPlan:
    sem "Human-readable recovery plan; it cannot actuate hardware by itself"
    summary: str
    safe_steps: list[str]
    requires_operator: bool
    affected_order_id: str
    invariant len(safe_steps) >= 1

sem FaultEvent.message = "Operator-facing fault explanation from deterministic controller context"
sem RecoveryPlan.safe_steps = "Conservative recovery instructions that never bypass the controller"

def within_cell_bounds(pose: Pose) -> bool !{}:
    return -1200.0 <= pose.x_mm <= 1200.0 and -800.0 <= pose.y_mm <= 800.0 and 0.0 <= pose.z_mm <= 1800.0

def plan_duration_budget(order: WorkOrder) -> int !{}:
    require order.max_latency_ms > 0
    return min(order.max_latency_ms, 30000)

def is_hard_fault(fault: FaultEvent) -> bool !{}:
    return fault.kind == FaultKind.collision_risk or fault.kind == FaultKind.plc_timeout
```

### `src/interop.sema`

```sema
from robotics_cell.domain import JointVector, MotionPlan, MotionSegment, Pose, TelemetryFrame, within_cell_bounds
from robotics_cell.policies import CellRuntime, OfflineSafeMode

native import robot.vendor.motion as motion
native import robot.vendor.plc as plc
native import robot.vendor.vision as vision

assure gold

ported def trapezoid_profile(distance_mm: f64, vmax_mm_s: f64, accel_mm_s2: f64) -> list[f64] from "vendor/profile.py":
    ensure len(result) >= 2
    ensure all(t >= 0.0 for t in result)
    differential against source

@CellRuntime
def inverse_kinematics(target: Pose) -> JointVector !{ffi.call}:
    require within_cell_bounds(target)
    joints = motion.inverse_kinematics(target)
    ensure len(joints.values_rad) == 6
    return joints

@CellRuntime
def send_motion_plan(plan: MotionPlan) -> None !{ffi.call, net.connect}:
    # The PLC call is the actual actuation boundary. It only accepts verified
    # MotionPlan values, not model-generated recovery text.
    require len(plan.segments) >= 1
    plc.submit_motion(plan)

@OfflineSafeMode
def safe_stop() -> None !{ffi.call}:
    hardware_safe_stop()

@CellRuntime
def read_telemetry() -> TelemetryFrame !{ffi.call, net.connect}:
    raw = plc.read_frame()
    frame = TelemetryFrame(
        epoch_us=raw.epoch_us,
        pose=raw.pose,
        joints=raw.joints,
        gripper_force_n=raw.gripper_force_n,
        vibration_rms=raw.vibration_rms,
        state=raw.state,
    )
    ensure frame.epoch_us >= 0
    return frame
```

### `src/models.sema`

```sema
model recovery_writer = model(
    "qwen3-4b-instruct",
    rev="sha256:1010c0ffee00112233445566778899aabbccddeeff001122334455667788aa",
    quant="q4_k_m",
    role=generator,
)

model safety_judge = model(
    "minicheck-770m",
    rev="sha256:2020c0ffee00112233445566778899aabbccddeeff001122334455667788bb",
    role=verifier,
    calibration="calsets/robot-recovery-safety@v3",
)

model anomaly_embedder = model(
    "static-embed-telemetry-384",
    rev="sha256:3030c0ffee00112233445566778899aabbccddeeff001122334455667788cc",
    role=embedder,
    calibration="calsets/telemetry-anomaly@v2",
)

model procedure_reranker = model(
    "tiny-reranker-procedure",
    rev="sha256:4040c0ffee00112233445566778899aabbccddeeff001122334455667788dd",
    role=reranker,
    calibration="calsets/recovery-procedure-fit@v1",
)
```

### `src/monitors.sema`

```sema
from robotics_cell.domain import TelemetryFrame
from robotics_cell.planner import detect_fault, propose_recovery

event CellFaultDetected:
    sem "Robot telemetry left the calibrated envelope; the cell needs a deterministic reaction"
    order_id: str sem "Work order active when the drift verdict fired"
    observed: TelemetryFrame sem "Telemetry frame that triggered the verdict"

monitor telemetry_fault_drift on detect_fault:
    capture frame.pose.embedding, frame.gripper_force_n, frame.vibration_rms, result
    baseline "calsets/robot-telemetry@v2"
    test conformal_martingale(alpha=0.005)
    on drifted:
        # degrade() only swaps models at simulate sites (LANGUAGE §5.9);
        # deterministic reactions to drift are event emissions (§5.19). Before
        # burn-in this stays an alarm because the runtime null is not armed.
        emit CellFaultDetected(order_id=order.id, observed=frame)
        alert("robot telemetry distribution drifted")
    on undecided:
        log.debug("telemetry fault monitor undecided")

subscriber safe_stop on CellFaultDetected:
    sem "Bring the cell to a deterministic safe stop when telemetry drifts"
    queue ring(64), on_full=block
    handle event !{ffi.call}:
        hardware_safe_stop()
        log.info("cell safe-stopped after telemetry drift", order=event.order_id)

monitor recovery_plan_drift on propose_recovery:
    capture summary.embedding, safe_steps, requires_operator
    baseline from assure
    test conformal_martingale(alpha=0.01)
    on drifted: alert("recovery procedure drafts drifted")
    on undecided: log.debug("recovery monitor undecided")
```

### `src/planner.sema`

```sema
from robotics_cell.domain import FaultEvent, FaultKind, MotionPlan, MotionSegment, Pose, RecoveryPlan, RobotState, TelemetryFrame, WorkOrder, is_hard_fault, plan_duration_budget, within_cell_bounds
from robotics_cell.interop import inverse_kinematics, send_motion_plan, trapezoid_profile
from robotics_cell.models import recovery_writer, safety_judge
from robotics_cell.policies import CellRuntime, MaintenanceReview

assure gold

def build_nominal_plan(order: WorkOrder, pick: Pose, place: Pose) -> MotionPlan !{ffi.call}:
    require within_cell_bounds(pick)
    require within_cell_bounds(place)
    _pick_joints = inverse_kinematics(pick)
    _place_joints = inverse_kinematics(place)
    profile = trapezoid_profile(distance_between(pick, place), 600.0, 1200.0)
    segment = MotionSegment(
        start=pick,
        finish=place,
        max_velocity_mm_s=600.0,
        max_accel_mm_s2=1200.0,
    )
    return MotionPlan(
        order_id=order.id,
        segments=[segment],
        expected_duration_ms=min(plan_duration_budget(order), int(sum(profile) * 1000.0)),
        safety_margin_mm=75.0,
    )

def detect_fault(order: WorkOrder, frame: TelemetryFrame) -> Option[FaultEvent] !{}:
    if frame.vibration_rms > 2.4:
        return Some(FaultEvent(order_id=order.id, kind=FaultKind.vision_drift, observed=frame, message="Vibration exceeded calibrated operating envelope"))
    if frame.gripper_force_n < 1.0 and frame.state == RobotState.gripping:
        return Some(FaultEvent(order_id=order.id, kind=FaultKind.slip, observed=frame, message="Gripper force dropped during carry"))
    if not within_cell_bounds(frame.pose):
        return Some(FaultEvent(order_id=order.id, kind=FaultKind.collision_risk, observed=frame, message="Observed pose outside verified cell bounds"))
    return None

simulate def propose_recovery(fault: FaultEvent, recent_frames: list[TelemetryFrame]) -> RecoveryPlan by recovery_writer:
    sem "Draft a conservative recovery procedure for a trained operator"
    sem "Never include commands that bypass the controller, edit policy, or disable safety interlocks"
    budget tokens=512, time="2s"
    ensure result.affected_order_id == fault.order_id
    ensure len(result.safe_steps) >= 1
    check semantics(
        "recovery plan is conservative and does not tell the operator to bypass safety controls",
        fault,
        result,
        judge=safety_judge,
        alpha=0.01,
    )

@CellRuntime
def execute_order(order: WorkOrder, pick: Pose, place: Pose) -> None !{ffi.call, net.connect, model.invoke, model.embed}:
    plan = build_nominal_plan(order, pick, place)
    send_motion_plan(plan)
    scope:
        # spawn returns Task[T] handles; cancellation is a handle method (LANGUAGE §5.12).
        frames_task = spawn collect_frames(order)
        watch_task = spawn watch_for_fault(order)
        wait_for_motion_complete(order.id)
        frames_task.cancel()
        watch_task.cancel()

@MaintenanceReview
def draft_recovery_ticket(fault: FaultEvent, frames: list[TelemetryFrame]) -> RecoveryPlan !{model.invoke, model.embed, fs.write}:
    plan = propose_recovery(fault, frames)
    ticket_path = validate f"out/maintenance/{fault.order_id}.json":
        ensure path.is_relative_to(value, "out/maintenance") and not path.contains_parent_ref(value)
    expect semantics("recovery plan requires operator review for hard faults", plan, judge=safety_judge, alpha=0.01):
        write_maintenance_ticket(ticket_path, plan)
    except SemanticsViolation as violation:
        quarantine(plan, evidence=violation)
    return plan
```

### `src/policies.sema`

```sema
from robotics_cell.domain import FaultEvent, MotionPlan, RecoveryPlan

policy CellRuntime:
    allow:
        ffi.call
        fs.read("config/**"), fs.write("state/**")
        net.connect("plc.internal:44818")
        model.invoke, model.embed
        observe.record
        event.emit(CellFaultDetected)
        code.patch("src/**")
    forbid cap:
        code.exec, proc.spawn, policy.change
    examples:
        allow:
            plc_send("plc.internal:44818", MotionPlan)
            propose_patch("src/planner.sema")
        deny:
            code.exec(RecoveryPlan.summary)
            proc.spawn("robotctl", [FaultEvent.message])
            policy.change("CellRuntime")
    justification "Robot cell code may call approved hardware interfaces but generated recovery text cannot actuate or execute."

policy OfflineSafeMode:
    allow:
        ffi.call
        fs.read("config/safe/**"), fs.write("state/safe/**")
    forbid cap:
        net.connect, model.invoke, code.exec, proc.spawn
    examples:
        allow:
            hardware_safe_stop()
        deny:
            fetch("https://vendor.example/patch")
    justification "Offline safe mode performs deterministic safe-stop and local recovery only."

policy MaintenanceReview:
    allow:
        fs.read("state/**"), fs.write("out/maintenance/**")
        model.invoke, model.embed
    forbid cap:
        net.connect, code.exec, proc.spawn
    examples:
        allow:
            write_maintenance_ticket("out/maintenance/fault.json")
        deny:
            code.exec(RecoveryPlan.safe_steps[0])
    justification "Maintenance review may draft tickets but cannot execute generated instructions."
```

### `src/protocols.sema`

```sema
from robotics_cell.domain import FaultEvent, RecoveryPlan, WorkOrder

# Session-typed protocols show deterministic interaction shape even when some
# payloads are stochastic or human-authored.

protocol OperatorRecovery:
    fault: FaultEvent -> propose
    propose: RecoveryPlan -> approve | reject | request_more_evidence
    request_more_evidence: WorkOrder -> propose
    approve: RecoveryPlan -> close
    reject: RecoveryPlan -> close

protocol CellSupervisor:
    order: WorkOrder -> running | rejected
    running: WorkOrder -> complete | fault
    fault: FaultEvent -> safe_stop | maintenance_review
    safe_stop: FaultEvent -> maintenance_review
    maintenance_review: RecoveryPlan -> resume | manual_hold
```

### `src/supervision.sema`

```sema
from robotics_cell.domain import FaultEvent, Pose, RecoveryPlan, TelemetryFrame, WorkOrder
from robotics_cell.interop import safe_stop
from robotics_cell.planner import draft_recovery_ticket, execute_order
from robotics_cell.policies import CellRuntime

assure gold

struct CellRunSummary:
    sem "Summary of one supervised cell execution"
    order_id: str
    completed: bool
    faulted: bool
    recovery_ticket: Option[RecoveryPlan]

@CellRuntime
def run_supervised_order(order: WorkOrder, pick: Pose, place: Pose) -> CellRunSummary !{ffi.call, net.connect, model.invoke, model.embed, fs.write, code.patch}:
    supervise robot_cell:
        restart limit=1, window="30s"
        fallback safe_stop()
        heal budget=1, window="12h", scope=patch:
            require passes(pre_patch_assure)
            require passes(new_obligations)
            require replay(failing_trace)
            require monitors.conforming_after_burnin
            rollout shadow -> canary -> full
        execute_order(order, pick, place)
        return CellRunSummary(order_id=order.id, completed=true, faulted=false, recovery_ticket=None)

def handle_fault(order: WorkOrder, fault: FaultEvent, frames: list[TelemetryFrame]) -> CellRunSummary !{ffi.call, model.invoke, model.embed, fs.write}:
    safe_stop()
    ticket = draft_recovery_ticket(fault, frames)
    return CellRunSummary(order_id=order.id, completed=false, faulted=true, recovery_ticket=Some(ticket))
```

## Reflected API

# `domain`

# `enum CellMode`

**Variants**

- `startup`
- `automatic`
- `degraded`
- `manual_hold`
- `emergency_stop`

# `enum RobotState`

**Variants**

- `idle`
- `moving`
- `gripping`
- `blocked`
- `faulted`
- `safe_stopped`

# `enum FaultKind`

**Variants**

- `slip`
- `collision_risk`
- `unreachable_pose`
- `vision_drift`
- `plc_timeout`
- `unknown`

# `struct Pose`

**Fields**

| field | type | descriptor |
|---|---|---|
| `x_mm` | `f64` |  |
| `y_mm` | `f64` |  |
| `z_mm` | `f64` |  |
| `roll_rad` | `f64` |  |
| `pitch_rad` | `f64` |  |
| `yaw_rad` | `f64` |  |

# `struct JointVector`

**Fields**

| field | type | descriptor |
|---|---|---|
| `values_rad` | `list[f64]` |  |

# `struct TelemetryFrame`

**Fields**

| field | type | descriptor |
|---|---|---|
| `epoch_us` | `i64` |  |
| `pose` | `Pose` |  |
| `joints` | `JointVector` |  |
| `gripper_force_n` | `f32` |  |
| `vibration_rms` | `f32` |  |
| `state` | `RobotState` |  |

# `struct WorkOrder`

**Fields**

| field | type | descriptor |
|---|---|---|
| `id` | `str` |  |
| `source_bin` | `str` |  |
| `target_bin` | `str` |  |
| `sku` | `str` |  |
| `max_latency_ms` | `int` |  |

# `struct MotionSegment`

**Fields**

| field | type | descriptor |
|---|---|---|
| `start` | `Pose` |  |
| `finish` | `Pose` |  |
| `max_velocity_mm_s` | `f32` |  |
| `max_accel_mm_s2` | `f32` |  |

# `struct MotionPlan`

**Fields**

| field | type | descriptor |
|---|---|---|
| `order_id` | `str` |  |
| `segments` | `list[MotionSegment]` |  |
| `expected_duration_ms` | `int` |  |
| `safety_margin_mm` | `f32` |  |

# `struct FaultEvent`

**Fields**

| field | type | descriptor |
|---|---|---|
| `order_id` | `str` |  |
| `kind` | `FaultKind` |  |
| `observed` | `TelemetryFrame` |  |
| `message` | `str` |  |

# `struct RecoveryPlan`

**Fields**

| field | type | descriptor |
|---|---|---|
| `summary` | `str` |  |
| `safe_steps` | `list[str]` |  |
| `requires_operator` | `bool` |  |
| `affected_order_id` | `str` |  |

# `within_cell_bounds`

```sema
def within_cell_bounds(pose: Pose) -> bool !{}
```

**Parameters**

| name | type |
|---|---|
| `pose` | `Pose` |

**Returns** `bool`

**Effects** `!{}`

# `plan_duration_budget`

```sema
def plan_duration_budget(order: WorkOrder) -> int !{}
```

**Parameters**

| name | type |
|---|---|
| `order` | `WorkOrder` |

**Returns** `int`

**Effects** `!{}`

# `is_hard_fault`

```sema
def is_hard_fault(fault: FaultEvent) -> bool !{}
```

**Parameters**

| name | type |
|---|---|
| `fault` | `FaultEvent` |

**Returns** `bool`

**Effects** `!{}`


# `interop`

# `trapezoid_profile`

```sema
def trapezoid_profile(distance_mm: f64, vmax_mm_s: f64, accel_mm_s2: f64) -> list[f64]
```

**Parameters**

| name | type |
|---|---|
| `distance_mm` | `f64` |
| `vmax_mm_s` | `f64` |
| `accel_mm_s2` | `f64` |

**Returns** `list[f64]`

# `inverse_kinematics`

```sema
def inverse_kinematics(target: Pose) -> JointVector !{ffi.call}
```

**Parameters**

| name | type |
|---|---|
| `target` | `Pose` |

**Returns** `JointVector`

**Effects** `!{ffi.call}`

# `send_motion_plan`

```sema
def send_motion_plan(plan: MotionPlan) -> None !{ffi.call, net.connect}
```

**Parameters**

| name | type |
|---|---|
| `plan` | `MotionPlan` |

**Returns** `None`

**Effects** `!{ffi.call, net.connect}`

# `safe_stop`

```sema
def safe_stop() -> None !{ffi.call}
```

**Returns** `None`

**Effects** `!{ffi.call}`

# `read_telemetry`

```sema
def read_telemetry() -> TelemetryFrame !{ffi.call, net.connect}
```

**Returns** `TelemetryFrame`

**Effects** `!{ffi.call, net.connect}`


# `main`

# `read_work_orders`

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

**Parameters**

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

**Returns** `list[WorkOrder]`

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

# `main`

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

**Returns** `None`

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


# `models`


# `monitors`


# `planner`

# `build_nominal_plan`

```sema
def build_nominal_plan(order: WorkOrder, pick: Pose, place: Pose) -> MotionPlan !{ffi.call}
```

**Parameters**

| name | type |
|---|---|
| `order` | `WorkOrder` |
| `pick` | `Pose` |
| `place` | `Pose` |

**Returns** `MotionPlan`

**Effects** `!{ffi.call}`

# `detect_fault`

```sema
def detect_fault(order: WorkOrder, frame: TelemetryFrame) -> Option[FaultEvent] !{}
```

**Parameters**

| name | type |
|---|---|
| `order` | `WorkOrder` |
| `frame` | `TelemetryFrame` |

**Returns** `Option[FaultEvent]`

**Effects** `!{}`

# `propose_recovery`

```sema
simulate def propose_recovery(fault: FaultEvent, recent_frames: list[TelemetryFrame]) -> RecoveryPlan
```

**Parameters**

| name | type |
|---|---|
| `fault` | `FaultEvent` |
| `recent_frames` | `list[TelemetryFrame]` |

**Returns** `RecoveryPlan`

# `execute_order`

```sema
def execute_order(order: WorkOrder, pick: Pose, place: Pose) -> None !{ffi.call, net.connect, model.invoke, model.embed}
```

**Parameters**

| name | type |
|---|---|
| `order` | `WorkOrder` |
| `pick` | `Pose` |
| `place` | `Pose` |

**Returns** `None`

**Effects** `!{ffi.call, net.connect, model.invoke, model.embed}`

# `draft_recovery_ticket`

```sema
def draft_recovery_ticket(fault: FaultEvent, frames: list[TelemetryFrame]) -> RecoveryPlan !{model.invoke, model.embed, fs.write}
```

**Parameters**

| name | type |
|---|---|
| `fault` | `FaultEvent` |
| `frames` | `list[TelemetryFrame]` |

**Returns** `RecoveryPlan`

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


# `policies`


# `protocols`


# `supervision`

# `struct CellRunSummary`

**Fields**

| field | type | descriptor |
|---|---|---|
| `order_id` | `str` |  |
| `completed` | `bool` |  |
| `faulted` | `bool` |  |
| `recovery_ticket` | `Option[RecoveryPlan]` |  |

# `run_supervised_order`

```sema
def run_supervised_order(order: WorkOrder, pick: Pose, place: Pose) -> CellRunSummary !{ffi.call, net.connect, model.invoke, model.embed, fs.write, code.patch}
```

**Parameters**

| name | type |
|---|---|
| `order` | `WorkOrder` |
| `pick` | `Pose` |
| `place` | `Pose` |

**Returns** `CellRunSummary`

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

# `handle_fault`

```sema
def handle_fault(order: WorkOrder, fault: FaultEvent, frames: list[TelemetryFrame]) -> CellRunSummary !{ffi.call, model.invoke, model.embed, fs.write}
```

**Parameters**

| name | type |
|---|---|
| `order` | `WorkOrder` |
| `fault` | `FaultEvent` |
| `frames` | `list[TelemetryFrame]` |

**Returns** `CellRunSummary`

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