Steering

The steering inbox and event types that let external callers cancel or redirect an in-flight planner run.

steering

Steering event models and inbox for bidirectional control.

SteeringInbox

SteeringInbox(*, maxsize: int = 100, max_pending_user_messages: int = 2)

Async inbox that buffers steering events and exposes cancellation state.

A running task (typically the ReactPlanner loop) polls or awaits this inbox to react to externally published :class:~penguiflow.state.models.SteeringEvent records — cancelling, pausing/resuming, or draining events to fold into the next planning step. CANCEL and PAUSE/RESUME events are handled specially at :meth:push time so their effects (cancelled, wait_if_paused) are visible even before the event is drained from the queue.

Initialize the inbox.

Parameters:

Name	Type	Description	Default
maxsize	int	Maximum number of buffered steering events before :meth:push starts returning False.	100
max_pending_user_messages	int	Maximum number of undrained USER_MESSAGE events allowed in the queue at once; additional ones are rejected by :meth:push.	2

cancelled property

cancelled: bool

Whether a CANCEL steering event has been pushed to this inbox.

cancel_reason property

cancel_reason: str | None

The reason string from the most recent CANCEL event, if any was pushed.

cancel_event property

cancel_event: Event

The underlying :class:asyncio.Event set when a CANCEL event is pushed.

push async

push(event: SteeringEvent) -> bool

Queue a steering event, returning False if the queue is full.

CANCEL events set :attr:cancel_event and record :attr:cancel_reason; PAUSE/RESUME events immediately toggle the pause gate used by :meth:wait_if_paused — both take effect even if the queue itself is full. USER_MESSAGE events are additionally capped by max_pending_user_messages.

Parameters:

Name	Type	Description	Default
event	SteeringEvent	The steering event to enqueue.	required

Returns:

Type	Description
bool	True if the event was enqueued; False if the queue was full or the
bool	USER_MESSAGE backlog limit was reached.

has_event

has_event() -> bool

Check if there are queued steering events without draining them.

Returns:

Type	Description
bool	True if at least one event is currently buffered.

drain

drain() -> list[SteeringEvent]

Drain any queued steering events without blocking.

Returns:

Type	Description
list[SteeringEvent]	All currently buffered events, in FIFO order; an empty list if none are queued.

next async

next() -> SteeringEvent

Wait for the next steering event.

Returns:

Type	Description
SteeringEvent	The next :class:~penguiflow.state.models.SteeringEvent pushed to the inbox.

wait_if_paused async

wait_if_paused() -> None

Block until a RESUME arrives if the task is paused.

No-op if the inbox is not currently paused (i.e. no PAUSE event is outstanding).

SteeringEvent

Bases: BaseModel

session_id instance-attribute

session_id: str

task_id instance-attribute

task_id: str

event_id class-attribute instance-attribute

event_id: str = Field(default_factory=lambda: uuid.uuid4().hex)

event_type instance-attribute

event_type: SteeringEventType

payload class-attribute instance-attribute

payload: dict[str, Any] = Field(default_factory=dict)

trace_id class-attribute instance-attribute

trace_id: str | None = None

source class-attribute instance-attribute

source: str = 'user'

created_at class-attribute instance-attribute

created_at: datetime = Field(default_factory=_utc_now)

to_injection

to_injection() -> str

SteeringEventType

Bases: str, Enum

INJECT_CONTEXT class-attribute instance-attribute

INJECT_CONTEXT = 'INJECT_CONTEXT'

REDIRECT class-attribute instance-attribute

REDIRECT = 'REDIRECT'

CANCEL class-attribute instance-attribute

CANCEL = 'CANCEL'

PRIORITIZE class-attribute instance-attribute

PRIORITIZE = 'PRIORITIZE'

PAUSE class-attribute instance-attribute

PAUSE = 'PAUSE'

RESUME class-attribute instance-attribute

RESUME = 'RESUME'

APPROVE class-attribute instance-attribute

APPROVE = 'APPROVE'

REJECT class-attribute instance-attribute

REJECT = 'REJECT'

USER_MESSAGE class-attribute instance-attribute

USER_MESSAGE = 'USER_MESSAGE'

SteeringCancelled

SteeringCancelled(reason: str | None = None)

Bases: RuntimeError

Raised when a steering cancel event terminates a task.

Attributes:

Name	Type	Description
reason		Human-readable reason for the cancellation (defaults to "steering_cancelled" when none was provided).

Initialize the exception with an optional cancellation reason.

Parameters:

Name	Type	Description	Default
reason	str | None	Human-readable reason for the cancellation, taken from the CANCEL steering event's payload when available.	None

reason instance-attribute

reason = reason or 'steering_cancelled'