robotics-cell

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

Run it from sema/:

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

Source

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

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

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

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

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

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

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

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

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

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

Parameters

name	type
pose	Pose

Returns bool

Effects !{}

plan_duration_budget

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

Parameters

name	type
order	WorkOrder

Returns int

Effects !{}

is_hard_fault

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

Parameters

name	type
fault	FaultEvent

Returns bool

Effects !{}

interop

trapezoid_profile

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

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

Parameters

name	type
target	Pose

Returns JointVector

Effects !{ffi.call}

send_motion_plan

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

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

Returns None

Effects !{ffi.call}

read_telemetry

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

Returns TelemetryFrame

Effects !{ffi.call, net.connect}

main

read_work_orders

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

Parameters

name	type
path	str

Returns list[WorkOrder]

Effects !{fs.read}

main

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

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

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

Parameters

name	type
order	WorkOrder
frame	TelemetryFrame

Returns Option[FaultEvent]

Effects !{}

propose_recovery

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

Parameters

name	type
fault	FaultEvent
recent_frames	list[TelemetryFrame]

Returns RecoveryPlan

execute_order

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

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

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

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}