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

test "cell bounds include the envelope and reject every escaped axis":
    ensure within_cell_bounds(Pose(x_mm=-1200.0, y_mm=-800.0, z_mm=0.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure within_cell_bounds(Pose(x_mm=1200.0, y_mm=800.0, z_mm=1800.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure not within_cell_bounds(Pose(x_mm=-1200.1, y_mm=0.0, z_mm=1.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure not within_cell_bounds(Pose(x_mm=1200.1, y_mm=0.0, z_mm=1.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure not within_cell_bounds(Pose(x_mm=0.0, y_mm=-800.1, z_mm=1.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure not within_cell_bounds(Pose(x_mm=0.0, y_mm=800.1, z_mm=1.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure not within_cell_bounds(Pose(x_mm=0.0, y_mm=0.0, z_mm=-0.1, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))
    ensure not within_cell_bounds(Pose(x_mm=0.0, y_mm=0.0, z_mm=1800.1, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0))

test "duration budgeting preserves small deadlines and caps large ones":
    short = WorkOrder(id="short", source_bin="a", target_bin="b", sku="s", max_latency_ms=1)
    long = WorkOrder(id="long", source_bin="a", target_bin="b", sku="s", max_latency_ms=45000)
    ensure plan_duration_budget(short) == 1
    ensure plan_duration_budget(long) == 30000

test "only collision and PLC timeout are hard faults":
    pose = Pose(x_mm=0.0, y_mm=0.0, z_mm=100.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0)
    frame = TelemetryFrame(epoch_us=1, pose=pose, joints=JointVector(values_rad=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0]), gripper_force_n=10.0, vibration_rms=0.1, state=RobotState.idle)
    ensure is_hard_fault(FaultEvent(order_id="o", kind=FaultKind.collision_risk, observed=frame, message="outside bounds"))
    ensure is_hard_fault(FaultEvent(order_id="o", kind=FaultKind.plc_timeout, observed=frame, message="PLC timeout"))
    ensure not is_hard_fault(FaultEvent(order_id="o", kind=FaultKind.slip, observed=frame, message="slip"))

src/interop.sema

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

import math

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

assure gold

def trapezoid_profile(distance_mm: f64, vmax_mm_s: f64, accel_mm_s2: f64) -> list[f64] !{}:
    require distance_mm >= 0.0
    require vmax_mm_s > 0.0 and accel_mm_s2 > 0.0
    ensure len(result) == 3 or len(result) == 4
    ensure result[0] == 0.0
    ensure all(t >= 0.0 for t in result)
    ensure all(result[i] <= result[i + 1] for i in range(len(result) - 1))
    accel_time = vmax_mm_s / accel_mm_s2
    accel_distance = 0.5 * accel_mm_s2 * accel_time ** 2
    if 2.0 * accel_distance >= distance_mm:
        peak_time = math.sqrt(distance_mm / accel_mm_s2)
        return [0.0, peak_time, 2.0 * peak_time]
    cruise_time = (distance_mm - 2.0 * accel_distance) / vmax_mm_s
    return [0.0, accel_time, accel_time + cruise_time, 2.0 * accel_time + cruise_time]

@CellRuntime
def inverse_kinematics(target: Pose) -> JointVector !{ffi.call}:
    require within_cell_bounds(target)
    return validate_joint_vector(motion.inverse_kinematics(target))

def validate_joint_vector(joints: JointVector) -> JointVector !{}:
    require 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,
    )
    return validate_telemetry_frame(frame)

def validate_telemetry_frame(frame: TelemetryFrame) -> TelemetryFrame !{}:
    require frame.epoch_us >= 0
    return frame

test "native joint and telemetry fixtures cross validators unchanged":
    pose = Pose(x_mm=10.0, y_mm=20.0, z_mm=30.0, roll_rad=0.1, pitch_rad=0.2, yaw_rad=0.3)
    joints = JointVector(values_rad=[0.0, 0.1, 0.2, 0.3, 0.4, 0.5])
    frame = TelemetryFrame(epoch_us=7, pose=pose, joints=joints, gripper_force_n=12.0, vibration_rms=0.2, state=RobotState.moving)
    checked_joints = validate_joint_vector(joints)
    checked_frame = validate_telemetry_frame(frame)
    ensure checked_joints.values_rad == joints.values_rad
    ensure checked_frame.epoch_us == 7
    ensure checked_frame.pose == pose
    ensure checked_frame.state == RobotState.moving

test "motion profile selects bounded triangular and trapezoidal regimes":
    triangular = trapezoid_profile(100.0, 600.0, 1200.0)
    trapezoidal = trapezoid_profile(1200.0, 600.0, 1200.0)
    ensure len(triangular) == 3
    ensure abs(triangular[-1] - 0.5773502691896257) < 0.000000001
    ensure trapezoidal == [0.0, 0.5, 2.0, 2.5]

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

import math

assure gold

def distance_between(left: Pose, right: Pose) -> f64 !{}:
    ensure result >= 0.0
    dx = right.x_mm - left.x_mm
    dy = right.y_mm - left.y_mm
    dz = right.z_mm - left.z_mm
    return math.sqrt(dx ** 2 + dy ** 2 + dz ** 2)

def build_nominal_plan(order: WorkOrder, pick: Pose, place: Pose) -> MotionPlan !{}:
    require within_cell_bounds(pick)
    require within_cell_bounds(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 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 FaultEvent(order_id=order.id, kind=FaultKind.slip, observed=frame, message="Gripper force dropped during carry")
    if not within_cell_bounds(frame.pose):
        return FaultEvent(order_id=order.id, kind=FaultKind.collision_risk, observed=frame, message="Observed pose outside verified cell bounds")
    return None

test "fault detection is deterministic and priority ordered":
    order = WorkOrder(id="order-9", source_bin="a", target_bin="b", sku="fixture", max_latency_ms=1000)
    pose = Pose(x_mm=0.0, y_mm=0.0, z_mm=100.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0)
    joints = JointVector(values_rad=[0.0, 0.0, 0.0, 0.0, 0.0, 0.0])
    nominal = TelemetryFrame(epoch_us=1, pose=pose, joints=joints, gripper_force_n=8.0, vibration_rms=0.1, state=RobotState.idle)
    vibration = TelemetryFrame(epoch_us=2, pose=pose, joints=joints, gripper_force_n=0.1, vibration_rms=2.5, state=RobotState.gripping)
    slip = TelemetryFrame(epoch_us=3, pose=pose, joints=joints, gripper_force_n=0.5, vibration_rms=0.1, state=RobotState.gripping)
    escaped = TelemetryFrame(epoch_us=4, pose=Pose(x_mm=1200.1, y_mm=0.0, z_mm=100.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0), joints=joints, gripper_force_n=8.0, vibration_rms=0.1, state=RobotState.moving)
    vibration_boundary = TelemetryFrame(epoch_us=5, pose=pose, joints=joints, gripper_force_n=8.0, vibration_rms=2.4, state=RobotState.idle)
    force_boundary = TelemetryFrame(epoch_us=6, pose=pose, joints=joints, gripper_force_n=1.0, vibration_rms=0.1, state=RobotState.gripping)
    match detect_fault(order, nominal):
        case Some(_):
            ensure false
        case None:
            ensure true
    match detect_fault(order, vibration):
        case Some(fault):
            ensure fault.kind == FaultKind.vision_drift
            ensure fault.message == "Vibration exceeded calibrated operating envelope"
        case None:
            ensure false
    match detect_fault(order, slip):
        case Some(fault):
            ensure fault.kind == FaultKind.slip
        case None:
            ensure false
    match detect_fault(order, escaped):
        case Some(fault):
            ensure fault.kind == FaultKind.collision_risk
        case None:
            ensure false
    match detect_fault(order, vibration_boundary):
        case Some(_):
            ensure false
        case None:
            ensure true
    match detect_fault(order, force_boundary):
        case Some(_):
            ensure false
        case None:
            ensure true

test "nominal planning preserves motion data and exercises both duration bounds":
    pick = Pose(x_mm=0.0, y_mm=0.0, z_mm=100.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0)
    place = Pose(x_mm=0.0, y_mm=0.0, z_mm=200.0, roll_rad=0.0, pitch_rad=0.0, yaw_rad=0.0)
    high_budget = WorkOrder(id="high-budget", source_bin="a", target_bin="b", sku="fixture", max_latency_ms=1000)
    low_budget = WorkOrder(id="low-budget", source_bin="a", target_bin="b", sku="fixture", max_latency_ms=500)
    high_plan = build_nominal_plan(high_budget, pick, place)
    low_plan = build_nominal_plan(low_budget, pick, place)
    ensure distance_between(pick, place) == 100.0
    ensure distance_between(pick, place) == distance_between(place, pick)
    ensure high_plan.order_id == "high-budget"
    ensure high_plan.segments == [MotionSegment(start=pick, finish=place, max_velocity_mm_s=600.0, max_accel_mm_s2=1200.0)]
    ensure high_plan.expected_duration_ms == 866
    ensure high_plan.safety_margin_mm == 75.0
    ensure low_plan.order_id == "low-budget"
    ensure low_plan.segments == high_plan.segments
    ensure low_plan.expected_duration_ms == 500
    ensure low_plan.safety_margin_mm == 75.0

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}:
    # Keep provider-backed reachability validation at the actuation boundary;
    # deterministic plan construction remains independently verifiable.
    _pick_joints = inverse_kinematics(pick)
    _place_joints = inverse_kinematics(place)
    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]

def cell_recovery_invariants_hold() -> bool !{}:
    # Real pre-acceptance obligation: a completed summary must never carry a
    # fault or a recovery ticket before any patch is trusted.
    probe = completed_summary("gate-probe")
    return probe.completed and not probe.faulted

def failed_order_replays_fixed() -> bool !{}:
    # Gate closed until a real replay harness exists — the patch stays
    # rejected and the cell recovers via safe_stop.
    return false

@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
        fallback safe_stop()
        heal budget=1:
            # Acceptance gates are ordinary user predicates (LANGUAGE §5.11):
            # each is evaluated and journaled as decision:heal.gate.
            require cell_recovery_invariants_hold()
            require failed_order_replays_fixed()
            rollout shadow -> canary -> full
        execute_order(order, pick, place)
        return completed_summary(order.id)

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 fault_summary(order.id, ticket)

def completed_summary(order_id: str) -> CellRunSummary !{}:
    return CellRunSummary(order_id=order_id, completed=true, faulted=false, recovery_ticket=None)

def fault_summary(order_id: str, ticket: RecoveryPlan) -> CellRunSummary !{}:
    return CellRunSummary(order_id=order_id, completed=false, faulted=true, recovery_ticket=ticket)

test "supervision summaries distinguish completion from operator recovery":
    ticket = RecoveryPlan(summary="inspect gripper", safe_steps=["safe stop", "inspect"], requires_operator=true, affected_order_id="order-9")
    completed = completed_summary("order-8")
    faulted = fault_summary("order-9", ticket)
    ensure completed.order_id == "order-8"
    ensure completed.completed
    ensure not completed.faulted
    match completed.recovery_ticket:
        case Some(_):
            ensure false
        case None:
            ensure true
    ensure faulted.order_id == "order-9"
    ensure not faulted.completed
    ensure faulted.faulted
    match faulted.recovery_ticket:
        case Some(recovery):
            ensure recovery.summary == "inspect gripper"
            ensure recovery.affected_order_id == "order-9"
        case None:
            ensure false

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

def within_cell_bounds

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

Parameters

name	type
pose	Pose

Returns bool

Effects !{}

def plan_duration_budget

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

Parameters

name	type
order	WorkOrder

Returns int

Effects !{}

def is_hard_fault

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

Parameters

name	type
fault	FaultEvent

Returns bool

Effects !{}

interop

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

Effects !{}

def inverse_kinematics

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

Parameters

name	type
target	Pose

Returns JointVector

Effects !{ffi.call}

def validate_joint_vector

def validate_joint_vector(joints: JointVector) -> JointVector !{}

Parameters

name	type
joints	JointVector

Returns JointVector

Effects !{}

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

def safe_stop

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

Returns None

Effects !{ffi.call}

def read_telemetry

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

Returns TelemetryFrame

Effects !{ffi.call, net.connect}

def validate_telemetry_frame

def validate_telemetry_frame(frame: TelemetryFrame) -> TelemetryFrame !{}

Parameters

name	type
frame	TelemetryFrame

Returns TelemetryFrame

Effects !{}

main

def read_work_orders

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

Parameters

name	type
path	str

Returns list[WorkOrder]

Effects !{fs.read}

def 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

def distance_between

def distance_between(left: Pose, right: Pose) -> f64 !{}

Parameters

name	type
left	Pose
right	Pose

Returns f64

Effects !{}

def build_nominal_plan

def build_nominal_plan(order: WorkOrder, pick: Pose, place: Pose) -> MotionPlan !{}

Parameters

name	type
order	WorkOrder
pick	Pose
place	Pose

Returns MotionPlan

Effects !{}

def detect_fault

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

Parameters

name	type
order	WorkOrder
frame	TelemetryFrame

Returns Option[FaultEvent]

Effects !{}

def propose_recovery

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

Parameters

name	type
fault	FaultEvent
recent_frames	list[TelemetryFrame]

Returns RecoveryPlan

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

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

def cell_recovery_invariants_hold

def cell_recovery_invariants_hold() -> bool !{}

Returns bool

Effects !{}

def failed_order_replays_fixed

def failed_order_replays_fixed() -> bool !{}

Returns bool

Effects !{}

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

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

def completed_summary

def completed_summary(order_id: str) -> CellRunSummary !{}

Parameters

name	type
order_id	str

Returns CellRunSummary

Effects !{}

def fault_summary

def fault_summary(order_id: str, ticket: RecoveryPlan) -> CellRunSummary !{}

Parameters

name	type
order_id	str
ticket	RecoveryPlan

Returns CellRunSummary

Effects !{}