step.background_manager

Background step manager — tracks running background tasks and handles preemption.

When a foreground step requires a resource held by a background step, the background step is cancelled (preempted) with a warning instead of raising ResourceConflictError. The manager also logs when background steps complete, fail, or get cancelled.

Attributes

logger

Classes

BackgroundManager

Track background step tasks and handle resource preemption.

Functions

get_background_manager(→ BackgroundManager)

Get or create the BackgroundManager attached to robot.

Module Contents

step.background_manager.logger

class step.background_manager.BackgroundManager

Track background step tasks and handle resource preemption.

Attached to the robot instance as robot._background_manager.

register(task: asyncio.Task[None], label: str, name: str | None, resources: frozenset[str]) → None

Register a background task for tracking and preemption.

async preempt_conflicts(resources: frozenset[str], holder_label: str) → None

Cancel background tasks whose resources overlap with resources.

After cancellation, awaits cleanup so released resources are available for immediate acquisition.

async wait_all() → None

Wait for all running background tasks to complete.

async wait_for_name(name: str) → None

Wait for a specific named background task to complete.

Either outcome (clean exit, cancellation, or exception inside the task) is acceptable here — the caller is asking “tell me when it’s done”, not “tell me how it ended”. We split the two except clauses on purpose: CancelledError is a BaseException in Python 3.8+, so the previous except (CancelledError, Exception) tuple was misleading. Task-internal failures get a debug log so a diagnostic trail exists; cancellation does not, because it is the expected outcome of mission shutdown.

async cancel_all() → None

Cancel all running background tasks and await cleanup.

Called at mission boundaries and during shutdown to prevent orphaned tasks from leaking or triggering asyncio warnings.

property active_count: int

Number of background tasks still running.

step.background_manager.get_background_manager(robot: object) → BackgroundManager

Get or create the BackgroundManager attached to robot.