Simulation Control

Pause, resume, step, and breakpoint support for interactive simulation debugging.

Simulation control: pause/resume, stepping, breakpoints, and introspection.

Breakpoint

Bases: Protocol

Protocol for breakpoint implementations.

Breakpoints are evaluated after each event. Implement should_break to define the trigger condition. Set one_shot to True for breakpoints that should auto-remove after triggering once.

ConditionBreakpoint dataclass

ConditionBreakpoint(
    fn: Callable[[BreakpointContext], bool],
    description: str = "custom condition",
    one_shot: bool = False,
)

Pause when a user-defined predicate returns True.

Parameters:

Name	Type	Description	Default
fn	Callable[[BreakpointContext], bool]	Callable receiving BreakpointContext, returns bool.	required
description	str	Human-readable description for listing breakpoints.	'custom condition'
one_shot	bool	If True, removed after first trigger. Defaults to False.	False

EventCountBreakpoint dataclass

EventCountBreakpoint(count: int, one_shot: bool = True)

Pause after a given number of events have been processed.

Parameters:

Name	Type	Description	Default
count	int	The event count threshold.	required
one_shot	bool	If True (default), removed after first trigger.	True

EventTypeBreakpoint dataclass

EventTypeBreakpoint(
    event_type: str, one_shot: bool = False
)

Pause when the last processed event matches a specific type.

Parameters:

Name	Type	Description	Default
event_type	str	The event_type string to match.	required
one_shot	bool	If True, removed after first trigger. Defaults to False.	False

MetricBreakpoint dataclass

MetricBreakpoint(
    entity_name: str,
    attribute: str,
    operator: str,
    threshold: float,
    one_shot: bool = False,
)

Pause when an entity attribute crosses a threshold.

Looks up the named entity in the simulation, reads the given attribute, and applies the comparison operator against the threshold.

Parameters:

Name	Type	Description	Default
entity_name	str	Name of the entity to inspect.	required
attribute	str	Attribute name to read from the entity.	required
operator	str	Comparison operator: "gt", "ge", "lt", "le", "eq", "ne".	required
threshold	float	Value to compare against.	required
one_shot	bool	If True, removed after first trigger. Defaults to False.	False

Raises:

Type	Description
ValueError	If the operator string is not recognized.

TimeBreakpoint dataclass

TimeBreakpoint(time: Instant, one_shot: bool = True)

Pause when simulation time reaches or exceeds a threshold.

Parameters:

Name	Type	Description	Default
time	Instant	The simulation time at which to break.	required
one_shot	bool	If True (default), removed after first trigger.	True

SimulationControl

SimulationControl(simulation: Simulation)

Interactive control surface for a running simulation.

Created lazily via sim.control. Provides:

• Execution control: pause, resume, step, reset
• Breakpoints: condition-based pausing (time, count, metric, etc.)
• Event hooks: callbacks on event processing and time advance
• Heap introspection: peek at upcoming events without consuming them

Parameters:

Name	Type	Description	Default
simulation	Simulation	The simulation instance to control.	required

is_paused property

is_paused: bool

True if the simulation is currently paused.

is_running property

is_running: bool

True if the simulation has started and has not completed.

pause

pause() -> None

Request the simulation to pause before processing the next event.

Takes effect at the next control check point in the event loop.

resume

resume()

Resume a paused simulation.

Returns:

Name	Type	Description
SimulationSummary		Partial summary if paused again, or
		final summary if the simulation completes.

step

step(n: int = 1)

Process exactly n events then pause.

Parameters:

Name	Type	Description	Default
n	int	Number of events to process. Must be >= 1.	1

Returns:

Name	Type	Description
SimulationSummary		Partial summary after stepping.

get_state

get_state() -> SimulationState

Return a snapshot of the simulation's current state.

reset

reset() -> None

Reset the simulation to its initial state.

Clears the event heap, resets the clock and counters, and re-primes sources and probes. Does NOT reset entity internal state.

add_breakpoint

add_breakpoint(bp: Breakpoint) -> str

Register a breakpoint. Returns a unique ID for removal.

remove_breakpoint

remove_breakpoint(bp_id: str) -> None

Remove a breakpoint by its ID.

Raises:

Type	Description
KeyError	If the ID is not found.

list_breakpoints

list_breakpoints() -> list[tuple[str, Breakpoint]]

Return all registered breakpoints as (id, breakpoint) pairs.

clear_breakpoints

clear_breakpoints() -> None

Remove all breakpoints.

on_event

on_event(callback: Callable[[Event], None]) -> str

Register a hook called after each event is processed.

Parameters:

Name	Type	Description	Default
callback	Callable[[Event], None]	Function receiving the processed Event.	required

Returns:

Type	Description
str	Hook ID for later removal.

on_time_advance

on_time_advance(callback: Callable[[Instant], None]) -> str

Register a hook called when simulation time advances.

Parameters:

Name	Type	Description	Default
callback	Callable[[Instant], None]	Function receiving the new simulation time.	required

Returns:

Type	Description
str	Hook ID for later removal.

remove_hook

remove_hook(hook_id: str) -> None

Remove an event or time hook by its ID.

Raises:

Type	Description
KeyError	If the ID is not found in either hook registry.

peek_next

peek_next(n: int = 1) -> list[Event]

Preview the next n events without removing them from the heap.

Only available when the simulation is paused.

Parameters:

Name	Type	Description	Default
n	int	Number of events to preview.	1

Returns:

Type	Description
list[Event]	List of up to n events in scheduled order.

find_events

find_events(
    predicate: Callable[[Event], bool],
) -> list[Event]

Find all heap events matching a predicate.

Only available when the simulation is paused. Performs a linear scan.

Parameters:

Name	Type	Description	Default
predicate	Callable[[Event], bool]	Function returning True for matching events.	required

Returns:

Type	Description
list[Event]	List of matching events (unsorted).

BreakpointContext dataclass

BreakpointContext(
    current_time: Instant,
    events_processed: int,
    last_event: Event,
    simulation: Simulation,
)

Context passed to breakpoint predicates for evaluation.

Provides read-only access to simulation state so breakpoints can make decisions based on current time, event count, the last event processed, and entity attributes.

Attributes:

Name	Type	Description
current_time	Instant	Current simulation time.
events_processed	int	Total events processed so far.
last_event	Event	The event that was just processed.
simulation	Simulation	Reference to the simulation for entity inspection.

SimulationState dataclass

SimulationState(
    current_time: Instant,
    events_processed: int,
    heap_size: int,
    primary_events_remaining: int,
    is_paused: bool,
    is_running: bool,
    is_complete: bool,
    last_event: Event | None,
    wall_clock_elapsed: float,
)

Immutable snapshot of the simulation's current status.

Returned by SimulationControl.get_state() for inspection without modifying simulation internals.

Attributes:

Name	Type	Description
current_time	Instant	The simulation clock's current time.
events_processed	int	Total events processed so far.
heap_size	int	Total events remaining in the event heap.
primary_events_remaining	int	Non-daemon events remaining.
is_paused	bool	True if execution was explicitly paused.
is_running	bool	True if the simulation has started and not completed.
is_complete	bool	True if the simulation has finished.
last_event	Event | None	The most recently processed event, or None.
wall_clock_elapsed	float	Real-time seconds since the simulation started.