step.watchdog_manager

Watchdog manager — deadlines that cancel the main-mission task on expiry.

Two usage patterns share one manager:

• Mission deadlines — armed automatically by the robot runner from a Mission.time_budget class attribute. Never fed; expiry fires when the budget elapses.

• Keepalive watchdogs — armed/fed/disarmed imperatively by user code via the start_watchdog / feed_watchdog / stop_watchdog steps. Must be fed before their timeout or expiry fires.

On expiry, the manager cancels the registered main-mission task. This routes through the existing shutdown path in _run_missions: background tasks drain, the shutdown mission runs, then the process exits.

Attributes

logger

Exceptions

WatchdogExpiredError

Raised on the main-mission task when a watchdog expires.

Classes

WatchdogManager

Track armed watchdogs and cancel the main-mission task on expiry.

Functions

get_watchdog_manager(→ WatchdogManager)

Get or create the WatchdogManager attached to robot.

Module Contents

step.watchdog_manager.logger

exception step.watchdog_manager.WatchdogExpiredError

Bases: Exception

Raised on the main-mission task when a watchdog expires.

class step.watchdog_manager.WatchdogManager

Track armed watchdogs and cancel the main-mission task on expiry.

Attached to the robot instance as robot._watchdog_manager. Wire the main-mission task via attach_main_task before arming anything so the manager knows what to cancel.

attach_main_task(task: asyncio.Task[None]) → None

Register the task that should be cancelled when a watchdog fires.

detach_main_task() → None

Clear the main-task reference after main missions finish.

property expired_name: str | None

Name of the watchdog that fired, or None if no expiry occurred.

arm(name: str, timeout: float, source: str = 'user') → None

Arm a watchdog. Replaces any existing entry with the same name.

feed(name: str) → None

Push the named watchdog’s deadline out by its full timeout.

disarm(name: str, *, missing_ok: bool = False) → None

Cancel and remove the named watchdog.

active_names(source: str | None = None) → list[str]

Return names of currently armed watchdogs, optionally filtered by source.

async cancel_all() → None

Cancel every armed watchdog and await cleanup.

step.watchdog_manager.get_watchdog_manager(robot: object) → WatchdogManager

Get or create the WatchdogManager attached to robot.