01 | EVERY ACTION NEEDS AN EXIT

A drive command can keep requesting movement even when a wheel is blocked. A sensor read can return slowly. A match can change mode while an autonomous action is still running.

Success is only one possible exit. A routine must also handle cancellation, expired time budgets, invalid feedback, missed updates and insufficient measured progress.

This guide extends 09.1. The example drives 300 mm, turns clockwise 90° and pauses, with checks around each stage and before output.

02 | DEADLINES AND WATCHDOGS ANSWER DIFFERENT QUESTIONS

A fast loop can repeatedly process bad or unchanged data. A moving encoder can belong to a spinning wheel that is not moving the chassis. Choose evidence that matches the failure you want to detect.

03 | CAPTURE A DEADLINE ONCE

started = brain.timer.time(MSEC)
deadline = started + timeout_ms
# On every update:
if now >= deadline:
    return "FAILED", "ACTION_TIMEOUT"

Set the deadline when entering the action, not every time the loop runs. Continually replacing deadline with now + timeout prevents expiry.

Use elapsed time or a fixed absolute deadline from the same timer. Do not reset the Brain timer during the routine. The examples assume it remains monotonic over the short run.

Expiry means failure to finish within budget. It is not success and must not silently start the next normal action.

04 | SHARE ONE ROUTE BUDGET

Individual action limits do not constrain the total route. Eight seconds of driving, six of turning and a pause can exceed a fourteen-second route budget.

The earliest applicable deadline wins. Sensor acquisition, settling, transitions and pauses all consume the shared route budget. Leave measured room for stopping and uncertainty.

A fallback must use the remaining route budget. Do not create a fresh full allowance after every failure or retry.

05 | A HEARTBEAT MEANS COMPLETED WORK

A heartbeat is a timestamp or sequence number recording completed work. Here, completed_at is renewed only after the checks and both motor-output calls have completed.

# A loop entry alone is not proof that its work completed.
if now - completed_at > MAX_UPDATE_GAP_MS:
    return "UPDATE_GAP"
# Renew only after the cycle completes.
completed_at = after_output

The runner also measures snapshot-call duration and rechecks elapsed time before and between output calls. A delayed call causes an overrun result when it returns.

Reading the current encoder value is not proof that the underlying device produced a new sample. This is an acquisition-timing check, not a sensor frame-counter guarantee.

06 | KNOW WHAT A SOFTWARE WATCHDOG CANNOT DO

A watchdog check in the same loop cannot execute while that loop is permanently blocked. If an API call never returns, the following check and finally block cannot run at that moment.

An independent supervisory task can sometimes observe a stalled worker, but it still depends on the runtime scheduling it. It is not equivalent to an independent hardware watchdog. This starter does not add a second motor-writing task.

Use short, non-blocking action updates and test actual field-control interruption. Do not rely on Python checks as a replacement for the platform’s disable behaviour or physical protection of a mechanism.

07 | WATCH MEASURED PROGRESS

The example monitors each side’s averaged motor encoder distance while the previous output request is at least 5% in magnitude. It expects at least 5 mm of encoder-derived displacement before 750 ms passes.

A low request, a direction change or an action transition resets the watch. This avoids demanding progress during deliberate braking, settling or a pause.

The magnitude check is intentionally limited: movement in the wrong direction can pass it. Action target checks still determine completion. Averaging also cannot identify every individual motor failure.

Wheel slip can produce encoder progress with little chassis motion. An inertial turn may still time out. For stronger evidence, combine tracking-wheel, heading or wall-range observations as appropriate.

08 | CANCEL, LATCH AND RESTART DELIBERATELY

Each callback run gets a RunSession identity. The runner checks that it still owns the active session, remains authorized and is still in enabled autonomous mode. Pressing B cancels the example.

Revocation is latched: once a session is revoked, it cannot authorize output again, even if B is released or mode changes back. The driver callback revokes the old session and attempts to brake.

A new autonomous callback is rejected while the previous callback is still active. After cleanup and exit, a new run creates a new machine and captures fresh starting feedback. It never resumes an old target using stale entry values.

The shared identity is a teaching ownership pattern, not a general thread-safe lock. Keep one motor owner and test the VEX callback lifecycle before extending this to independently scheduled writers.

09 | COMPLETE GUARDED AUTONOMOUS STARTER

Paste below # Begin project code, keeping generated VEX imports and Brain setup unchanged. Configure manually below that point or use Devices, without duplicate declarations.

Four-motor drive: left ports 3/4, right 1/2, blue 6:1 cartridges. Inertial: 20. Verify motor direction, encoder scale, sensor mounting and a clear practice path. MOTION_CONFIGURED starts False. The encoder scale and progress thresholds are illustrative.

# =============================================================================
# TEAM [MY TEAM NAME / NUMBER] | WATCHDOGS, DEADLINES AND INTERRUPTION
# =============================================================================
# Keep generated imports and brain above '# Begin project code'.
# Manual configuration below: do not duplicate declarations in Devices.
import math
left_1 = Motor(Ports.PORT3, GearSetting.RATIO_6_1, False)
left_2 = Motor(Ports.PORT4, GearSetting.RATIO_6_1, False)
right_1 = Motor(Ports.PORT1, GearSetting.RATIO_6_1, True)
right_2 = Motor(Ports.PORT2, GearSetting.RATIO_6_1, True)
left_drive = MotorGroup(left_1, left_2)
right_drive = MotorGroup(right_1, right_2)
inertial = Inertial(Ports.PORT20)
controller = Controller(PRIMARY)

MOTION_CONFIGURED = False
MM_PER_MOTOR_TURN = 219.44  # Example: calibrate wheel size/external gearing
DRIVE_MM = 300.0
TURN_DEG = 90.0            # Relative clockwise turn using continuous rotation
DRIVE_KP = 0.05
TURN_KP = 0.7
MAX_DRIVE = 20.0
MAX_TURN = 10.0
MAX_WHEEL_OUTPUT = 25.0
DISTANCE_TOL_MM = 10.0
HEADING_TOL_DEG = 2.0
SIDE_RATE_TOL_MM_S = 30.0
TURN_RATE_TOL_DEG_S = 8.0
SETTLE_MS = 200
DRIVE_TIMEOUT_MS = 8000
TURN_TIMEOUT_MS = 6000
PAUSE_MS = 500
ROUTE_TIMEOUT_MS = 14000
POLL_MS = 20
MAX_UPDATE_GAP_MS = 100
ready = False
PROGRESS_WINDOW_MS = 750
PROGRESS_MIN_MM = 5.0
PROGRESS_MIN_OUTPUT = 5.0
active_run = None
run_number = 0

def clamp(v, low, high):
    return max(low, min(high, v))

def angle_error(target, current):
    return (target-current+180)%360-180

def stop_drive():
    left_drive.stop(BRAKE)
    right_drive.stop(BRAKE)

def side_output(group, value):
    if value == 0:
        group.stop(BRAKE)
    else:
        group.spin(FORWARD if value > 0 else REVERSE, abs(value), PERCENT)

def read_snapshot():
    if not inertial.installed() or inertial.is_calibrating():
        raise ValueError("Inertial unavailable")
    turns = (left_1.position(TURNS), left_2.position(TURNS),
             right_1.position(TURNS), right_2.position(TURNS))
    rotation = inertial.rotation(DEGREES)
    if not all(math.isfinite(v) for v in turns+(rotation,)):
        raise ValueError("Invalid encoder/inertial data")
    return {"left": (turns[0]+turns[1])*MM_PER_MOTOR_TURN/2,
            "right": (turns[2]+turns[3])*MM_PER_MOTOR_TURN/2,
            "rotation": rotation}

# =============================================================================
# ACTION: NO LOOPS, WAITS OR DEVICE ACCESS INSIDE update()
# =============================================================================
class MotionAction:
    def __init__(self, mode, now, snapshot):
        if mode not in ("DRIVE", "TURN"):
            raise ValueError("Invalid action mode")
        self.mode = mode
        self.started = now
        self.start = dict(snapshot)  # Capture once; never reset on each update.
        self.settled_since = None
        self.finished = None

    def update(self, now, snapshot, slow):
        if self.finished is not None:
            return self.finished, 0.0, 0.0, ""
        limit = DRIVE_TIMEOUT_MS if self.mode == "DRIVE" else TURN_TIMEOUT_MS
        if now-self.started >= limit:
            self.finished = "FAILED"
            return "FAILED", 0.0, 0.0, self.mode+"_TIMEOUT"
        if self.mode == "DRIVE":
            left_distance = snapshot["left"]-self.start["left"]
            right_distance = snapshot["right"]-self.start["right"]
            distance_error = DRIVE_MM-(left_distance+right_distance)/2
            heading_error = angle_error(self.start["rotation"], snapshot["rotation"])
            if distance_error < -DISTANCE_TOL_MM:
                self.finished = "FAILED"
                return "FAILED", 0.0, 0.0, "DRIVE_OVERSHOOT"
            near = (abs(DRIVE_MM-left_distance) <= DISTANCE_TOL_MM
                    and abs(DRIVE_MM-right_distance) <= DISTANCE_TOL_MM
                    and abs(heading_error) <= HEADING_TOL_DEG)
            base = (clamp(DRIVE_KP*distance_error, 0, MAX_DRIVE)
                    if abs(heading_error) <= 10 else 0.0)
            turn = clamp(TURN_KP*heading_error, -MAX_TURN, MAX_TURN)
        else:
            # Relative target is fixed at entry; continuous rotation preserves +90.
            heading_error = self.start["rotation"]+TURN_DEG-snapshot["rotation"]
            near = abs(heading_error) <= HEADING_TOL_DEG
            base = 0.0
            turn = clamp(TURN_KP*heading_error, -MAX_TURN, MAX_TURN)
        if near:
            if slow:
                if self.settled_since is None:
                    self.settled_since = now
                if now-self.settled_since >= SETTLE_MS:
                    self.finished = "DONE"
                    return "DONE", 0.0, 0.0, ""
            else:
                self.settled_since = None
            return "RUNNING", 0.0, 0.0, ""  # Brake while observing settling.
        self.settled_since = None
        left, right = base+turn, base-turn
        scale = max(1.0, abs(left)/MAX_WHEEL_OUTPUT, abs(right)/MAX_WHEEL_OUTPUT)
        return "RUNNING", left/scale, right/scale, ""

# =============================================================================
# FINITE STATE MACHINE: DRIVE -> TURN -> PAUSE -> DONE; FAILURE -> FAULT
# =============================================================================
class AutoMachine:
    def __init__(self, now, snapshot):
        self.state = "DRIVE"
        self.reason = ""
        self.deadline = now+ROUTE_TIMEOUT_MS
        self.action = MotionAction("DRIVE", now, snapshot)
        self.pause_until = None

    def fail(self, reason):
        self.state, self.reason = "FAULT", reason
        return 0.0, 0.0

    def tick(self, now, snapshot, slow):
        if self.state in ("DONE", "FAULT"):
            return 0.0, 0.0
        if now >= self.deadline:
            return self.fail("ROUTE_DEADLINE")
        if self.state == "PAUSE":
            if now >= self.pause_until:
                self.state = "DONE"
            return 0.0, 0.0
        status, left, right, reason = self.action.update(now, snapshot, slow)
        if status == "FAILED":
            return self.fail(reason)
        if status == "DONE":
            if self.state == "DRIVE":
                self.state = "TURN"
                self.action = MotionAction("TURN", now, snapshot)  # Enter once.
            else:
                self.state = "PAUSE"
                self.pause_until = now+PAUSE_MS
                self.action = None
            return 0.0, 0.0  # Stop at the transition; next action starts next tick.
        return left, right

# =============================================================================
# PROGRESS WATCH: SIDE ENCODER MOVEMENT UNDER A SUSTAINED REQUEST
# =============================================================================
class ProgressWatch:
    def __init__(self):
        self.since = None
        self.position = None
        self.direction = 0
    def update(self, now, position, previous_output):
        direction = 1 if previous_output > 0 else -1
        if abs(previous_output) < PROGRESS_MIN_OUTPUT:
            self.since = self.position = None
            self.direction = 0
            return True
        if self.since is None or direction != self.direction:
            self.since, self.position, self.direction = now, position, direction
            return True
        if abs(position-self.position) >= PROGRESS_MIN_MM:
            self.since, self.position = now, position
            return True
        return now-self.since < PROGRESS_WINDOW_MS

# =============================================================================
# RUN IDENTITY AND LATCHED REVOCATION
# =============================================================================
class RunSession:
    def __init__(self, number):
        self.number = number
        self.revoked = False
    def revoke(self):
        self.revoked = True

def permitted(session):
    return (active_run is session and not session.revoked and ready
            and MOTION_CONFIGURED and competition.is_enabled()
            and competition.is_autonomous() and not controller.buttonB.pressing())

def stop_drive_best_effort():
    failed = False
    # A failed left stop must not prevent an attempt to stop the right side.
    for group in (left_drive, right_drive):
        try:
            group.stop(BRAKE)
        except Exception:
            failed = True
    return not failed

# =============================================================================
# GUARDED RUNNER: NO LONG WAIT OR NESTED MOTION LOOP
# =============================================================================
def guarded_loop(session):
    settings = (MM_PER_MOTOR_TURN, DRIVE_MM, TURN_DEG, DRIVE_KP, TURN_KP,
                MAX_DRIVE, MAX_TURN, MAX_WHEEL_OUTPUT, DISTANCE_TOL_MM,
                HEADING_TOL_DEG, SIDE_RATE_TOL_MM_S, TURN_RATE_TOL_DEG_S,
                SETTLE_MS, DRIVE_TIMEOUT_MS, TURN_TIMEOUT_MS, PAUSE_MS,
                ROUTE_TIMEOUT_MS, POLL_MS, MAX_UPDATE_GAP_MS)
    if (not all(math.isfinite(v) for v in settings)
            or min(MM_PER_MOTOR_TURN, DRIVE_MM, DRIVE_KP, TURN_KP,
                   MAX_DRIVE, MAX_TURN, MAX_WHEEL_OUTPUT, DISTANCE_TOL_MM,
                   HEADING_TOL_DEG, SETTLE_MS, DRIVE_TIMEOUT_MS,
                   TURN_TIMEOUT_MS, ROUTE_TIMEOUT_MS) <= 0
            or PAUSE_MS < 0 or SIDE_RATE_TOL_MM_S < 0 or TURN_RATE_TOL_DEG_S < 0
            or not 0 < POLL_MS < MAX_UPDATE_GAP_MS
            or max(MAX_DRIVE, MAX_TURN) > MAX_WHEEL_OUTPUT
            or MAX_WHEEL_OUTPUT > 25):
        return "INVALID_SETTINGS"
    if (not all(math.isfinite(x) for x in (PROGRESS_WINDOW_MS, PROGRESS_MIN_MM, PROGRESS_MIN_OUTPUT))
            or min(PROGRESS_WINDOW_MS, PROGRESS_MIN_MM, PROGRESS_MIN_OUTPUT) <= 0
            or PROGRESS_MIN_OUTPUT > MAX_WHEEL_OUTPUT):
        return "INVALID_PROGRESS_SETTINGS"
    if not permitted(session):
        return "CANCELLED_OR_NOT_READY"
    start = brain.timer.time(MSEC)
    route_deadline = start+ROUTE_TIMEOUT_MS  # Includes initial sensor acquisition.
    completed_at = start
    previous = None
    previous_ms = start
    previous_output = (0.0, 0.0)
    watches = (ProgressWatch(), ProgressWatch())
    machine = None
    while True:
        before = brain.timer.time(MSEC)
        if not permitted(session):
            return "CANCELLED_OR_NOT_READY"
        if before >= route_deadline:
            return "ROUTE_DEADLINE"
        if before-completed_at > MAX_UPDATE_GAP_MS:
            return "UPDATE_GAP"
        snapshot = read_snapshot()
        sampled_at = brain.timer.time(MSEC)
        if not permitted(session):
            return "CANCELLED_OR_NOT_READY"
        if sampled_at-before > MAX_UPDATE_GAP_MS:
            return "SENSOR_CALL_OVERRUN"
        if sampled_at-completed_at > MAX_UPDATE_GAP_MS:
            return "UPDATE_GAP"
        if sampled_at >= route_deadline:
            return "ROUTE_DEADLINE"
        dt = (sampled_at-previous_ms)/1000.0
        slow = (previous is not None and dt > 0 and
                max(abs(snapshot["left"]-previous["left"]),
                    abs(snapshot["right"]-previous["right"]))/dt <= SIDE_RATE_TOL_MM_S
                and abs(snapshot["rotation"]-previous["rotation"])/dt <= TURN_RATE_TOL_DEG_S)
        previous, previous_ms = snapshot, sampled_at
        if machine is None:
            machine = AutoMachine(sampled_at, snapshot)
            machine.deadline = route_deadline
        for watch, key, output in zip(watches, ("left", "right"), previous_output):
            if not watch.update(sampled_at, snapshot[key], output):
                return key.upper()+"_NO_ENCODER_PROGRESS"
        old_state = machine.state
        left, right = machine.tick(sampled_at, snapshot, slow)
        # Do not renew the heartbeat merely because the loop was entered.
        # Check the age of the acquired snapshot again before output.
        action_deadline = (machine.action.started + (DRIVE_TIMEOUT_MS
                if machine.action.mode == "DRIVE" else TURN_TIMEOUT_MS)
                if machine.action is not None and machine.state not in ("DONE", "FAULT")
                else route_deadline)
        commit_at = brain.timer.time(MSEC)
        if commit_at-sampled_at > MAX_UPDATE_GAP_MS:
            return "SNAPSHOT_AGE"
        if commit_at-completed_at > MAX_UPDATE_GAP_MS:
            return "UPDATE_GAP"
        if commit_at >= route_deadline:
            return "ROUTE_DEADLINE"
        if commit_at >= action_deadline:
            return "ACTION_DEADLINE_OVERRUN"
        if not permitted(session):
            return "CANCELLED_OR_NOT_READY"
        side_output(left_drive, left)
        # These are separate API calls, not an atomic transaction.
        if not permitted(session):
            return "CANCELLED_OR_NOT_READY"
        between = brain.timer.time(MSEC)
        if between-sampled_at > MAX_UPDATE_GAP_MS:
            return "OUTPUT_CALL_OVERRUN"
        if between-completed_at > MAX_UPDATE_GAP_MS:
            return "UPDATE_GAP"
        if between >= route_deadline:
            return "ROUTE_DEADLINE"
        if between >= action_deadline:
            return "ACTION_DEADLINE_OVERRUN"
        side_output(right_drive, right)
        after = brain.timer.time(MSEC)
        if not permitted(session):
            return "CANCELLED_OR_NOT_READY"
        if after-sampled_at > MAX_UPDATE_GAP_MS:
            return "OUTPUT_CALL_OVERRUN"
        if after-completed_at > MAX_UPDATE_GAP_MS:
            return "UPDATE_GAP"
        if after >= route_deadline:
            return "ROUTE_DEADLINE"
        if after >= action_deadline:
            return "ACTION_DEADLINE_OVERRUN"
        completed_at = after  # Heartbeat: all checks and both output calls completed.
        previous_output = (left, right)
        if machine.state != old_state:
            print("RUN", session.number, "STATE", old_state, "->", machine.state, machine.reason)
            watches = (ProgressWatch(), ProgressWatch())
        if machine.state in ("DONE", "FAULT"):
            return machine.state+": "+machine.reason
        wait(POLL_MS, MSEC)

def run_guarded(session):
    result = "NOT_STARTED"
    try:
        if not stop_drive_best_effort():
            result = "INITIAL_STOP_FAILED"
        else:
            result = guarded_loop(session)
    except Exception:
        result = "SENSOR_ACTION_OR_OUTPUT_EXCEPTION"
    finally:
        session.revoke()  # This run cannot authorize later output.
        cleaned = stop_drive_best_effort()
    if not cleaned:
        result += "; STOP_COMMAND_FAILED"
    return result

# =============================================================================
# CALLBACKS: REFUSE OVERLAP, CANCEL OLD RUN, START FROM FRESH FEEDBACK
# =============================================================================
def autonomous():
    global active_run, run_number
    if active_run is not None:
        print("RUN_REJECTED: previous callback has not exited")
        return
    run_number += 1
    session = RunSession(run_number)
    active_run = session
    try:
        print("RUN",session.number,run_guarded(session))
    finally:
        if active_run is session:
            active_run = None  # Release only after cleanup was attempted.

def driver_control():
    if active_run is not None:
        active_run.revoke()
    stop_drive_best_effort()  # Teaching project: no driver movement.

competition = Competition(driver_control, autonomous)
stop_drive_best_effort()
if MOTION_CONFIGURED and not competition.is_enabled() and inertial.installed():
    inertial.calibrate()  # Disabled, stationary setup; outside the timed routine.
    ready = not inertial.is_calibrating() and not competition.is_enabled()
while True:
    wait(100, MSEC)

The action code includes heading correction, distance/heading tolerances and low measured rates during settling. All deadlines are checked cooperatively. There is no worker thread that guarantees a stop while the interpreter or an API call remains blocked.

The driver callback only revokes and stops in this teaching project. It does not include joystick movement. Add driver code only with a tested ownership handover so an old routine’s cleanup cannot interrupt a new owner.

10 | CLEAN UP EVERY EXIT

The outer runner revokes the session and attempts to stop both drive groups in finally. If stopping one group raises an exception, it still attempts the other. A stop-command failure is included in the result.

finally helps when Python execution unwinds normally or through a caught exception. It does not guarantee execution after a forced task termination, power loss or a permanently blocked call.

Motor commands are sequential. Cancellation can occur between checks and calls, so they are not an atomic transaction. Rechecks reduce the window; they do not remove it.

For extra mechanisms, extend cleanup to every owned resource and choose appropriate stop modes. A gravity-loaded lift needs a mechanism-specific policy and travel protection.

11 | RECORD THE REASON

Log the run number, state, elapsed time, reason and key feedback. Avoid heavy printing or file writes in every 20 ms cycle; logging itself can cause timing faults.

12 | TEST INTERRUPTIONS WITHOUT CREATING A HAZARD

1. Use deterministic simulated feedback to check the normal sequence.
2. Keep encoder readings fixed in a mock and verify the progress fault.
3. Inject a finite call delay and verify rejection after it returns.
4. Inject sensor and output exceptions and inspect cleanup attempts.
5. Cancel before starting, during driving and during turning.
6. Change to disabled or driver mode during the routine.
7. Reject an overlapping callback, then start a fresh run after exit.
8. Measure a complete physical run and select justified timing limits.

Use simulation for blocked-progress and exception cases. Do not deliberately jam a physical drivetrain or insert a permanent blocking loop into a moving robot’s program.

13 | COMMON MISTAKES

• Renewing the deadline each tick: retain the entry deadline.
• Feeding a heartbeat before work completes: timestamp completed work.
• Calling an elapsed-time check a hardware watchdog: state its scheduling limits.
• Giving retries a new route budget: share the original deadline.
• Automatically resuming after interruption: revoke the old run and start fresh.
• Demanding progress during a stop: arm the watch only for a significant request.
• Calling encoder movement chassis progress: account for slip and feedback limitations.
• Assuming finally always stops the robot: distinguish cleanup attempts from hardware behaviour.

CHALLENGES | EXPIRE, INTERRUPT AND RESTART

1. Trace: calculate the earliest deadline in a drive-turn-pause route.
2. Mock: freeze the left-side feedback and identify the fault time.
3. Delay: inject a 150 ms sensor-call delay and explain when the check runs.
4. Cancel: press B during an action and prove release does not resume the old run.
5. Re-enter: reject a second callback until the first exits.
6. Restart: start another run at a different encoder position and verify fresh targets.
7. Clean up: make one mocked stop raise and verify an attempt on the other side.
8. Evaluate: explain why a permanently blocked worker needs a different protection layer.

OFFICIAL REFERENCES

• Competition callbacks and enabled/mode checks
• Motor output, position, timeout and stopping commands

Motor set_timeout applies to supported built-in movement operations. Do not treat it as a universal timeout for your own spin() loop or as proof that a target was achieved.

The run-session, watchdog and deadline code is independently written teaching code. It requires configuration, timing measurements and physical validation on your robot.