Sessions & scheduling

Session management, transports, task lifecycle objects, and the job scheduler for running background and recurring work.

session

Streaming session orchestration for bidirectional tasks.

SessionManager

SessionManager(*, limits: SessionLimits | None = None, state_store: SessionStateStore | StateStore | None = None, control_policy: ControlPolicy | None = None, telemetry_sink: TaskTelemetrySink | None = None)

Registry for StreamingSession instances by session_id.

get_or_create async

get_or_create(session_id: str) -> StreamingSession

Return the existing session for session_id, creating and hydrating one if needed.

Parameters:

Name	Type	Description	Default
session_id	str	The session identifier to look up or create.	required

Returns:

Type	Description
StreamingSession	The (possibly newly created) :class:StreamingSession, hydrated from the
StreamingSession	configured state store.

get async

get(session_id: str) -> StreamingSession | None

Return the session for session_id if it already exists, without creating one.

Parameters:

Name	Type	Description	Default
session_id	str	The session identifier to look up.	required

Returns:

Type	Description
StreamingSession | None	The existing :class:StreamingSession, or None if none has been created yet.

drop async

drop(session_id: str) -> None

Remove and close the session for session_id, if it exists.

Parameters:

Name	Type	Description	Default
session_id	str	The session identifier to remove.	required

SessionLimits dataclass

SessionLimits(max_tasks_per_session: int = 8, max_background_tasks: int = 4, max_concurrent_tasks: int = 3, max_task_runtime_s: float | None = 900, update_queue_size: int = 500, steering_queue_size: int = 200, max_pending_patches: int = 32, max_steering_payload_bytes: int = MAX_STEERING_PAYLOAD_BYTES, max_steering_events_per_task: int = 512)

Resource and safety limits enforced by a :class:StreamingSession.

Attributes:

Name	Type	Description
max_tasks_per_session	int	Maximum number of non-terminal tasks a session may hold at once.
max_background_tasks	int	Maximum number of non-terminal background tasks per session.
max_concurrent_tasks	int	Maximum number of tasks executing concurrently (via a semaphore); 0 disables the concurrency cap.
max_task_runtime_s	float | None	Wall-clock timeout applied to each task pipeline, or None for no limit.
update_queue_size	int	Maximum size of the update broker's per-subscriber queues.
steering_queue_size	int	Maximum size of each task's steering-event inbox.
max_pending_patches	int	Maximum number of context patches awaiting human approval at once.
max_steering_payload_bytes	int	Maximum allowed size of a steering event payload, in bytes.
max_steering_events_per_task	int	Maximum number of recently-seen event IDs tracked per task for duplicate detection.

max_tasks_per_session class-attribute instance-attribute

max_tasks_per_session: int = 8

max_background_tasks class-attribute instance-attribute

max_background_tasks: int = 4

max_concurrent_tasks class-attribute instance-attribute

max_concurrent_tasks: int = 3

max_task_runtime_s class-attribute instance-attribute

max_task_runtime_s: float | None = 900

update_queue_size class-attribute instance-attribute

update_queue_size: int = 500

steering_queue_size class-attribute instance-attribute

steering_queue_size: int = 200

max_pending_patches class-attribute instance-attribute

max_pending_patches: int = 32

max_steering_payload_bytes class-attribute instance-attribute

max_steering_payload_bytes: int = MAX_STEERING_PAYLOAD_BYTES

max_steering_events_per_task class-attribute instance-attribute

max_steering_events_per_task: int = 512

StreamingSession

StreamingSession(session_id: str, *, control_policy: ControlPolicy | None = None, limits: SessionLimits | None = None, state_store: SessionStateStore | StateStore | None = None, telemetry_sink: TaskTelemetrySink | None = None)

Manages bidirectional communication and task lifecycle.

A StreamingSession owns the tasks, steering inboxes, update broker, task groups, and shared conversation context for a single session_id. It is the primary runtime surface used by transports (connect), planner-facing meta-tools (spawning and steering tasks), and background-task orchestration (context patches, proactive reporting, and task groups). Instances are typically created and cached via :class:SessionManager rather than constructed directly.

session_id instance-attribute

session_id = session_id

registry property

registry: TaskRegistry

Return the :class:TaskRegistry tracking task state for this session.

limits property

limits: SessionLimits

Return the :class:SessionLimits configuration governing this session.

pending_patches property

pending_patches: dict[str, PendingContextPatch]

Return a snapshot copy of patches awaiting human approval, keyed by patch_id.

context_version property

context_version: int

Return the monotonically increasing version counter for the session context.

context_hash property

context_hash: str | None

Return the SHA-256 hash of the current llm_context, or None if unset.

get_context

get_context() -> tuple[dict[str, Any], dict[str, Any]]

Return copies of the session's current context dicts.

Returns:

Type	Description
dict[str, Any]	A (llm_context, tool_context) tuple of shallow copies, safe for callers to
dict[str, Any]	mutate without affecting session state.

get_background_results

get_background_results() -> dict[str, BackgroundTaskResult]

Return a copy of completed background task results, keyed by task_id.

Returns:

Type	Description
dict[str, BackgroundTaskResult]	A shallow copy of the session's background_results mapping.

configure_background_tasks

configure_background_tasks(config: BackgroundTasksConfig | None = None, **overrides: Any) -> BackgroundTasksConfig

Configure runtime background task behavior.

get_background_task_config

get_background_task_config() -> BackgroundTasksConfig

Return the active :class:BackgroundTasksConfig, creating a default one if unset.

Returns:

Type	Description
BackgroundTasksConfig	The current background-task configuration for this session.

set_foreground_busy

set_foreground_busy(*, busy: bool) -> None

Mark whether the foreground turn is currently busy or idle.

This gates the proactive reporter: reports are only generated while the foreground is idle.

Parameters:

Name	Type	Description	Default
busy	bool	True to mark the foreground as busy, False to mark it idle.	required

on_foreground_yield async

on_foreground_yield(*, turn_id: str | None = None) -> int

Handle the foreground yielding control, auto-sealing open task groups if configured.

Parameters:

Name	Type	Description	Default
turn_id	str | None	The turn whose open groups should be sealed. Defaults to the session's current turn (see set_turn_id) when omitted.	None

Returns:

Type	Description
int	The number of task groups sealed. Always 0 when
int	auto_seal_groups_on_foreground_yield is disabled in the background-task config.

suppress_proactive_reports

suppress_proactive_reports(*, task_ids: Iterable[str] | None = None, group_id: str | None = None) -> int

Suppress future proactive reports for the given tasks and/or group.

Once suppressed, matching task/group completions will not enqueue proactive report requests (e.g. because their results were already consumed by a retained turn).

Parameters:

Name	Type	Description	Default
task_ids	Iterable[str] | None	Task IDs to suppress individually.	None
group_id	str | None	A group ID to suppress as a whole.	None

Returns:

Type	Description
int	The number of newly-added suppressions (already-suppressed entries are not
int	counted again).

schedule_group_continuation

schedule_group_continuation(*, group_id: str, timeout_s: float | None, max_hops: int, cooldown_s: float) -> None

Schedule a background waiter that keeps polling a group after a retain-turn timeout.

Repeatedly waits for the group's completion (up to max_hops times, sleeping cooldown_s between attempts) so a group that timed out while retaining the turn is still tracked to completion. A no-op if a continuation for this group is already scheduled and running.

Parameters:

Name	Type	Description	Default
group_id	str	The task group to keep waiting on.	required
timeout_s	float | None	Per-attempt wait timeout in seconds, or None to wait indefinitely (in which case only one hop is meaningful).	required
max_hops	int	Maximum number of wait attempts; scheduling is skipped entirely if <= 0.	required
cooldown_s	float	Seconds to sleep between attempts when the previous attempt timed out.	required

connect async

connect(transport: Transport) -> SessionConnection

Bind a :class:Transport to this session.

Parameters:

Name	Type	Description	Default
transport	Transport	The transport implementation used to send/receive messages for this session.	required

Returns:

Type	Description
SessionConnection	A new :class:SessionConnection wrapping this session and the transport.

hydrate async

hydrate() -> None

Load persisted tasks from the state store into the in-memory registry.

Idempotent: subsequent calls after the first successful hydration are no-ops. Restores the most recent foreground task ID from the loaded tasks, if any.

update_context

update_context(*, llm_context: dict[str, Any] | None = None, tool_context: dict[str, Any] | None = None) -> None

Replace the session's llm_context and/or tool_context.

Any legacy inline background_results/background_result entries found in llm_context are extracted into the session's dedicated background-results store. Bumps context_version and recomputes context_hash whenever llm_context is provided.

Parameters:

Name	Type	Description	Default
llm_context	dict[str, Any] | None	Replacement dict for the LLM-facing context, or None to leave unchanged.	None
tool_context	dict[str, Any] | None	Replacement dict for the tool-facing context, or None to leave unchanged.	None

configure_proactive_reporting

configure_proactive_reporting(*, generator: Callable[[ProactiveReportRequest], Awaitable[None]], enabled: bool = False, strategies: list[str] | None = None, max_queued: int = 5, timeout_s: float = 30.0, max_hops: int = 2, fallback_notification: bool = True) -> None

Configure proactive report-back for background task completions.

Parameters:

Name	Type	Description	Default
generator	Callable[[ProactiveReportRequest], Awaitable[None]]	Async callable to generate proactive messages.	required
enabled	bool	Master switch for proactive reporting.	False
strategies	list[str] | None	Merge strategies that trigger reports (default: APPEND, REPLACE).	None
max_queued	int	Maximum queued reports before dropping oldest.	5
timeout_s	float	Timeout for proactive message generation.	30.0
max_hops	int	Maximum proactive recursion hops before disabling background spawning.	2
fallback_notification	bool	Fall back to notification panel if generation fails.	True

ensure_capacity async

ensure_capacity(task_type: TaskType) -> None

Verify the session has room for another task of the given type.

Parameters:

Name	Type	Description	Default
task_type	TaskType	The task type (FOREGROUND or BACKGROUND) to check capacity for.	required

Raises:

Type	Description
RuntimeError	With "task_limit_exceeded" if the session's active-task limit would be exceeded, or "background_task_limit_exceeded" if the background-task limit would be exceeded.

spawn_task async

spawn_task(pipeline: TaskPipeline, *, task_type: TaskType = TaskType.FOREGROUND, priority: int = 0, context_snapshot: TaskContextSnapshot | None = None, description: str | None = None, spawn_reason: str | None = None, parent_task_id: str | None = None, spawned_from_event_id: str | None = None, query: str | None = None, task_id: str | None = None, trace_id: str | None = None, merge_strategy: MergeStrategy = MergeStrategy.APPEND, propagate_on_cancel: Literal['cascade', 'isolate'] = 'cascade', notify_on_complete: bool = True, group_id: str | None = None) -> str

Create a task and run its pipeline concurrently, returning immediately.

Registers the task in the registry, builds (or adapts) its context snapshot, creates its steering inbox, and schedules pipeline on a background asyncio.Task without waiting for it to finish. Use this for background tasks, or for foreground tasks whose completion is awaited elsewhere.

Parameters:

Name	Type	Description	Default
pipeline	TaskPipeline	The async callable that executes the task given a :class:TaskRuntime.	required
task_type	TaskType	Whether this is a FOREGROUND or BACKGROUND task.	FOREGROUND
priority	int	Scheduling priority hint stored on the task.	0
context_snapshot	TaskContextSnapshot | None	An explicit context snapshot to use instead of building one from the session's current context.	None
description	str | None	Human-readable description of the task, used in summaries and notifications.	None
spawn_reason	str | None	Machine-readable reason the task was spawned (e.g. "tasks.spawn").	None
parent_task_id	str | None	The task that spawned this one, if any.	None
spawned_from_event_id	str | None	The steering/event ID that triggered this spawn, if any.	None
query	str | None	The user-facing query this task is answering, if applicable.	None
task_id	str | None	Explicit task ID to use; a random one is generated if omitted.	None
trace_id	str | None	Trace ID to associate with the task's updates.	None
merge_strategy	MergeStrategy	Strategy used to merge the task's resulting context patch.	APPEND
propagate_on_cancel	Literal['cascade', 'isolate']	Whether cancelling this task cascades to its children ("cascade") or leaves them running ("isolate").	'cascade'
notify_on_complete	bool	Whether a default completion notification is emitted when the task finishes without an explicit notification.	True
group_id	str | None	Task group to associate this task's completion with, if any.	None

Returns:

Type	Description
str	The task ID (either task_id or a newly generated one).

run_task async

run_task(pipeline: TaskPipeline, *, task_type: TaskType = TaskType.FOREGROUND, priority: int = 0, context_snapshot: TaskContextSnapshot | None = None, description: str | None = None, spawn_reason: str | None = None, parent_task_id: str | None = None, spawned_from_event_id: str | None = None, query: str | None = None, task_id: str | None = None, trace_id: str | None = None, merge_strategy: MergeStrategy = MergeStrategy.APPEND, propagate_on_cancel: Literal['cascade', 'isolate'] = 'cascade', notify_on_complete: bool = True) -> TaskResult

Create a task and run its pipeline to completion, awaiting the result.

Unlike :meth:spawn_task, this awaits the pipeline directly and returns its :class:TaskResult. When task_type is FOREGROUND, the foreground-busy flag is set for the duration of the call and on_foreground_yield is invoked once the pipeline finishes (success, failure, or cancellation).

Parameters:

Name	Type	Description	Default
pipeline	TaskPipeline	The async callable that executes the task given a :class:TaskRuntime.	required
task_type	TaskType	Whether this is a FOREGROUND or BACKGROUND task.	FOREGROUND
priority	int	Scheduling priority hint stored on the task.	0
context_snapshot	TaskContextSnapshot | None	An explicit context snapshot to use instead of building one from the session's current context.	None
description	str | None	Human-readable description of the task, used in summaries and notifications.	None
spawn_reason	str | None	Machine-readable reason the task was spawned.	None
parent_task_id	str | None	The task that spawned this one, if any.	None
spawned_from_event_id	str | None	The steering/event ID that triggered this spawn, if any.	None
query	str | None	The user-facing query this task is answering, if applicable.	None
task_id	str | None	Explicit task ID to use; a random one is generated if omitted.	None
trace_id	str | None	Trace ID to associate with the task's updates.	None
merge_strategy	MergeStrategy	Strategy used to merge the task's resulting context patch.	APPEND
propagate_on_cancel	Literal['cascade', 'isolate']	Whether cancelling this task cascades to its children ("cascade") or leaves them running ("isolate").	'cascade'
notify_on_complete	bool	Whether a default completion notification is emitted when the task finishes without an explicit notification.	True

Returns:

Name	Type	Description
The	TaskResult	class:TaskResult produced by the pipeline.

Raises:

Type	Description
RuntimeError	If session/task-type limits are exceeded (see ensure_capacity).
TimeoutError	If the pipeline exceeds limits.max_task_runtime_s.
SteeringCancelled	If the task is cancelled via steering while running.

steer async

steer(event: SteeringEvent) -> bool

Process an inbound steering event against this session's tasks.

Handles the full steering lifecycle: session/validation checks, duplicate-event detection, persistence, approve/reject routing for pending context patches and gated controls, control-policy confirmation gating (emitting a CHECKPOINT update when confirmation is required), and dispatch of PRIORITIZE, PAUSE, RESUME, and CANCEL (including cascading cancellation to child tasks). Finally, the event is pushed onto the target task's steering inbox if one exists.

Parameters:

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

Returns:

Type	Description
bool	True if the event was accepted/handled (including being pushed to the
bool	task's steering inbox); False if it was rejected, invalid, a duplicate,
bool	targeted a different session, or is pending confirmation.

get_task async

get_task(task_id: str) -> TaskState | None

Look up a task's current state by ID.

Parameters:

Name	Type	Description	Default
task_id	str	The task to look up.	required

Returns:

Type	Description
TaskState | None	The task's :class:TaskState, or None if no such task exists.

list_tasks async

list_tasks(*, status: TaskStatus | None = None) -> list[TaskState]

List tasks belonging to this session.

Parameters:

Name	Type	Description	Default
status	TaskStatus | None	If provided, only return tasks with this status.	None

Returns:

Type	Description
list[TaskState]	The matching list of :class:TaskState objects.

cancel_task async

cancel_task(task_id: str, *, reason: str | None = None) -> bool

Request cancellation of a task by sending a CANCEL steering event.

Parameters:

Name	Type	Description	Default
task_id	str	The task to cancel.	required
reason	str | None	Optional human-readable reason recorded on the cancellation.	None

Returns:

Type	Description
bool	True if the cancel event was accepted (see :meth:steer), False otherwise.

broadcast_steer async

broadcast_steer(*, event_type: SteeringEventType, payload: dict[str, Any] | None = None, source: str = 'user') -> int

Send the same steering event to every active task in this session.

Parameters:

Name	Type	Description	Default
event_type	SteeringEventType	The steering event type to broadcast (e.g. PAUSE, RESUME).	required
payload	dict[str, Any] | None	Payload to attach to each event.	None
source	str	The source recorded on each event (e.g. "user" or "system").	'user'

Returns:

Type	Description
int	The number of tasks that accepted the event (see :meth:steer).

list_updates async

list_updates(*, task_id: str | None = None, since_id: str | None = None, limit: int = 500) -> list[StateUpdate]

Fetch persisted state updates for this session from the state store.

Parameters:

Name	Type	Description	Default
task_id	str | None	If provided, only return updates for this task.	None
since_id	str | None	If provided, only return updates after this update ID.	None
limit	int	Maximum number of updates to return.	500

Returns:

Type	Description
list[StateUpdate]	The matching list of :class:StateUpdate objects, oldest constraints applied
list[StateUpdate]	by the underlying state store.

subscribe async

subscribe(*, task_ids: list[str] | None = None, update_types: list[UpdateType] | None = None, since_id: str | None = None) -> AsyncIterator[StateUpdate]

Subscribe to live state updates for this session, optionally replaying history.

Parameters:

Name	Type	Description	Default
task_ids	list[str] | None	If provided, only yield updates for these task IDs.	None
update_types	list[UpdateType] | None	If provided, only yield updates of these types.	None
since_id	str | None	If provided (and exactly one task_id is given), first replay persisted updates after this ID before switching to live updates.	None

Returns:

Type	Description
AsyncIterator[StateUpdate]	An async iterator of :class:StateUpdate objects matching the given filters.
AsyncIterator[StateUpdate]	The subscription is automatically released when the iterator is closed or
AsyncIterator[StateUpdate]	fully consumed.

apply_context_patch async

apply_context_patch(*, patch: ContextPatch, strategy: MergeStrategy = MergeStrategy.APPEND, payload: Any | None = None, group_id: str | None = None) -> str | None

Merge a completed task's context patch into the session, per strategy.

Detects whether the patch was produced from a context version/hash older than the session's current one (context_diverged) and emits a warning notification when so. For HUMAN_GATED, the patch is queued for approval and a CHECKPOINT update is published instead of being applied immediately. For APPEND/REPLACE, the patch is recorded as a background result immediately.

Parameters:

Name	Type	Description	Default
patch	ContextPatch	The context patch produced by the completed task.	required
strategy	MergeStrategy	The merge strategy to apply.	APPEND
payload	Any | None	The task's raw result payload, stored alongside the patch for APPEND/REPLACE strategies.	None
group_id	str | None	The task group this patch's task belongs to, if any.	None

Returns:

Type	Description
str | None	The generated patch_id if the patch is pending human approval
str | None	(HUMAN_GATED), otherwise None.

Raises:

Type	Description
RuntimeError	If the number of pending patches would exceed limits.max_pending_patches.

mark_background_consumed async

mark_background_consumed(*, task_ids: list[str]) -> int

Remove background task results from session context by task_id.

apply_pending_patch async

apply_pending_patch(*, patch_id: str, strategy: MergeStrategy | None = None) -> bool

Approve and merge a previously human-gated pending context patch.

Parameters:

Name	Type	Description	Default
patch_id	str	The ID of the pending patch to apply (see pending_patches).	required
strategy	MergeStrategy | None	Merge strategy to apply the patch with; defaults to APPEND.	None

Returns:

Type	Description
bool	True if a matching pending patch was found and applied, False if no
bool	pending patch exists for patch_id.

close async

close() -> None

Shut down the session, cancelling all in-flight tasks and background helpers.

Cancels the proactive reporter task, any scheduled group-continuation tasks, and every running task handle (marking each corresponding task as CANCELLED with error "session_closed"). Clears internal task-handle and steering-inbox tracking. Intended to be called once when a session is being torn down (see :meth:SessionManager.drop).

resolve_or_create_group async

resolve_or_create_group(*, group_name: str | None = None, group_id: str | None = None, turn_id: str | None = None, merge_strategy: MergeStrategy | None = None, report_strategy: GroupReportStrategy | None = None, retain_turn: bool = False) -> TaskGroup

Resolve an existing group or create a new one based on name/ID.

Turn-scoped name resolution: - If group_id provided, join that exact group (error if not found/joinable) - If group_name provided, resolve to OPEN group with that name created in the same turn, or create new group if none exists

add_task_to_group async

add_task_to_group(group_id: str, task_id: str) -> bool

Add a task to a group. Returns False if group not found or not open.

seal_group async

seal_group(group_id: str) -> bool

Seal a group (no more tasks can join). Returns False if already sealed.

get_group async

get_group(group_id: str | None = None, *, group_name: str | None = None, turn_id: str | None = None) -> TaskGroup | None

Get a task group by ID or by name+turn resolution.

Parameters:

Name	Type	Description	Default
group_id	str | None	Direct lookup by group ID	None
group_name	str | None	Name-based lookup (requires turn_id for disambiguation)	None
turn_id	str | None	Turn ID for name-based lookup	None

list_groups async

list_groups(*, status: GroupStatus | None = None) -> list[TaskGroup]

List all task groups, optionally filtered by status.

mark_task_complete_in_group async

mark_task_complete_in_group(group_id: str, task_id: str, *, success: bool, patch: ContextPatch | None = None, patch_id: str | None = None) -> None

Mark a task as complete (success or failure) within its group.

wait_for_group_completion async

wait_for_group_completion(group_id: str, *, timeout_s: float | None = None) -> tuple[TaskGroup | None, bool]

Wait for a group to complete.

Parameters:

Name	Type	Description	Default
group_id	str	The group to wait for.	required
timeout_s	float | None	Maximum time to wait in seconds. If None, wait indefinitely.	None

Returns:

Type	Description
TaskGroup | None	A tuple of (group, timed_out). If timed_out is True, the group may
bool	not be complete yet. If group is None, the group was not found.

get_group_results async

get_group_results(group_id: str) -> list[dict[str, Any]]

Get aggregated results/digests from a completed group's tasks.

cancel_group async

cancel_group(group_id: str, *, reason: str | None = None, propagate: bool = True) -> bool

Cancel a task group and optionally all its tasks.

apply_group_patches async

apply_group_patches(group_id: str, *, action: Literal['apply', 'reject'] = 'apply', strategy: MergeStrategy | None = None) -> bool

Apply or reject all pending patches for a task group.

set_turn_id

set_turn_id(turn_id: str | None) -> None

Set the current foreground turn ID for group name resolution.

auto_seal_open_groups async

auto_seal_open_groups(turn_id: str | None = None) -> int

Auto-seal all OPEN groups created in the given turn (or current turn).

Called when foreground yields to ensure groups don't stay open. Returns count of groups sealed.

TaskResult dataclass

TaskResult(payload: Any | None = None, context_patch: ContextPatch | None = None, digest: list[str] = list(), artifacts: list[dict[str, Any]] = list(), sources: list[dict[str, Any]] = list(), notification: NotificationPayload | None = None, metadata: dict[str, Any] = dict())

Outcome of a task pipeline, returned by :class:TaskPipeline callables.

Attributes:

Name	Type	Description
payload	Any | None	The task's primary output value, if any.
context_patch	ContextPatch | None	Optional patch to merge into the session context (see apply_context_patch).
digest	list[str]	Human-readable summary lines describing the result.
artifacts	list[dict[str, Any]]	Structured artifacts produced by the task (e.g. files, generated content).
sources	list[dict[str, Any]]	Structured source/citation references backing the result.
notification	NotificationPayload | None	Optional notification to surface to the user about this task's completion.
metadata	dict[str, Any]	Free-form metadata associated with the result.

payload class-attribute instance-attribute

payload: Any | None = None

context_patch class-attribute instance-attribute

context_patch: ContextPatch | None = None

digest class-attribute instance-attribute

digest: list[str] = field(default_factory=list)

artifacts class-attribute instance-attribute

artifacts: list[dict[str, Any]] = field(default_factory=list)

sources class-attribute instance-attribute

sources: list[dict[str, Any]] = field(default_factory=list)

notification class-attribute instance-attribute

notification: NotificationPayload | None = None

metadata class-attribute instance-attribute

metadata: dict[str, Any] = field(default_factory=dict)

TaskRuntime

TaskRuntime(*, session: StreamingSession, state: TaskState, steering: SteeringInbox, context_snapshot: TaskContextSnapshot)

Runtime helpers exposed to task pipelines.

session instance-attribute

session = session

state instance-attribute

state = state

steering instance-attribute

steering = steering

context_snapshot instance-attribute

context_snapshot = context_snapshot

emit_update

emit_update(update_type: UpdateType, content: Any, *, step_index: int | None = None, total_steps: int | None = None) -> StateUpdate

Build a :class:StateUpdate and publish it on the owning session.

If update_type is PROGRESS and content is a dict, the task's stored progress snapshot is also updated in the registry.

Parameters:

Name	Type	Description	Default
update_type	UpdateType	The kind of update being emitted (progress, result, etc.).	required
content	Any	The update payload; interpretation depends on update_type.	required
step_index	int | None	Optional 0-based index of the current step, for progress reporting.	None
total_steps	int | None	Optional total number of steps, for progress reporting.	None

Returns:

Name	Type	Description
The	StateUpdate	class:StateUpdate that was published.

notify

notify(payload: NotificationPayload) -> None

Emit a notification update for this task.

Parameters:

Name	Type	Description	Default
payload	NotificationPayload	The notification content (severity, title, body, actions) to publish.	required

transport

Transport contracts for bidirectional session connectivity.

SessionConnection

SessionConnection(session: StreamingSession, transport: Transport)

Wires a StreamingSession to a bidirectional transport.

Initialize the connection.

Parameters:

Name	Type	Description	Default
session	StreamingSession	The streaming session whose updates are forwarded and which receives steering events.	required
transport	Transport	The bidirectional transport to wire the session to.	required

Transport

Bases: Protocol

Bidirectional wire contract between a StreamingSession and a client.

Implementations bridge StateUpdates flowing out to the client and SteeringEvents flowing in from the client, over any concrete transport (WebSocket, SSE + POST, in-process queues, etc.).

send async

send(update: StateUpdate) -> None

Send a state update to the client.

Parameters:

Name	Type	Description	Default
update	StateUpdate	The state update to deliver.	required

receive async

receive() -> SteeringEvent | None

Wait for and return the next steering event from the client.

Returns:

Type	Description
SteeringEvent | None	The next SteeringEvent, or None if the transport has closed and no
SteeringEvent | None	more events will arrive.

close async

close() -> None

Close the transport and release any underlying resources.

scheduler

Scheduled job contracts and a lightweight scheduler loop.

JobScheduler

JobScheduler(*, store: JobStore, spawn: Callable[[JobDefinition], Awaitable[str]])

Polls due jobs and triggers task creation through a callback.

tick async

tick() -> None

Run one scheduling pass: spawn all due jobs and advance recurring schedules.

For each job returned by JobStore.list_due, records a JobRunRecord, invokes the spawn callback, and if the job repeats, computes and persists its next run time.

JobSchedulerRunner

JobSchedulerRunner(scheduler: JobScheduler, *, poll_interval_s: float = 5.0)

Background loop that ticks the scheduler on an interval.

start async

start() -> None

Start the background polling loop, if not already running.

Idempotent: calling start while a loop task is already active is a no-op.

stop async

stop() -> None

Cancel the background polling loop and wait for it to finish.

Safe to call when no loop is running.

JobDefinition

Bases: BaseModel

A scheduled background job: what to run, for which session, and when.

job_id class-attribute instance-attribute

job_id: str = Field(default_factory=lambda: secrets.token_hex(8), description='Unique identifier for this job.')

session_id class-attribute instance-attribute

session_id: str = Field(description='Session this job belongs to.')

task_payload class-attribute instance-attribute

task_payload: dict[str, Any] = Field(default_factory=dict, description='Payload passed to the spawn callback when the job fires.')

schedule class-attribute instance-attribute

schedule: ScheduleConfig = Field(description='Recurrence rule controlling when the job is due.')

delivery_policy class-attribute instance-attribute

delivery_policy: dict[str, Any] = Field(default_factory=dict, description='Policy controlling how results from this job are delivered.')

enabled class-attribute instance-attribute

enabled: bool = Field(default=True, description='Whether the job is eligible to run; disabled jobs are skipped.')

created_at class-attribute instance-attribute

created_at: datetime = Field(default_factory=_utc_now, description='When the job was created.')

updated_at class-attribute instance-attribute

updated_at: datetime = Field(default_factory=_utc_now, description='When the job was last modified.')

ScheduleConfig

Bases: BaseModel

Recurrence rule for a scheduled job.

A one-shot job sets only next_run_at; a recurring job also sets interval_s so next_after can compute the following run time after each tick.

interval_s class-attribute instance-attribute

interval_s: int | None = Field(default=None, description='Seconds between runs. None means the job does not repeat.')

next_run_at class-attribute instance-attribute

next_run_at: datetime | None = Field(default=None, description='Next timestamp at which the job is due to run.')

timezone class-attribute instance-attribute

timezone: str | None = Field(default=None, description='IANA timezone name used for interpreting schedule times, if any.')

next_after

next_after(when: datetime) -> datetime | None

Compute the next run time after a given timestamp.

Parameters:

Name	Type	Description	Default
when	datetime	The reference timestamp (typically the current run time) to offset from.	required

Returns:

Type	Description
datetime | None	The next run timestamp, or None if interval_s is not set (one-shot job).

models

Session/task models for bidirectional streaming and background work.

Most persistence-facing task/steering models live in penguiflow.state.models. This module re-exports them for backward compatibility.

TaskContextSnapshot

Bases: BaseModel

session_id instance-attribute

session_id: str

task_id instance-attribute

task_id: str

trace_id class-attribute instance-attribute

trace_id: str | None = None

spawned_from_task_id class-attribute instance-attribute

spawned_from_task_id: str = 'foreground'

spawned_from_event_id class-attribute instance-attribute

spawned_from_event_id: str | None = None

spawned_at class-attribute instance-attribute

spawned_at: datetime = Field(default_factory=_utc_now)

spawn_reason class-attribute instance-attribute

spawn_reason: str | None = None

query class-attribute instance-attribute

query: str | None = None

propagate_on_cancel class-attribute instance-attribute

propagate_on_cancel: str = 'cascade'

notify_on_complete class-attribute instance-attribute

notify_on_complete: bool = True

context_version class-attribute instance-attribute

context_version: int | None = None

context_hash class-attribute instance-attribute

context_hash: str | None = None

llm_context class-attribute instance-attribute

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

tool_context class-attribute instance-attribute

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

memory class-attribute instance-attribute

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

artifacts class-attribute instance-attribute

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

TaskStatus

Bases: str, Enum

PENDING class-attribute instance-attribute

PENDING = 'PENDING'

RUNNING class-attribute instance-attribute

RUNNING = 'RUNNING'

PAUSED class-attribute instance-attribute

PAUSED = 'PAUSED'

COMPLETE class-attribute instance-attribute

COMPLETE = 'COMPLETE'

FAILED class-attribute instance-attribute

FAILED = 'FAILED'

CANCELLED class-attribute instance-attribute

CANCELLED = 'CANCELLED'

TaskType

Bases: str, Enum

FOREGROUND class-attribute instance-attribute

FOREGROUND = 'FOREGROUND'

BACKGROUND class-attribute instance-attribute

BACKGROUND = 'BACKGROUND'

NotificationAction

Bases: BaseModel

A single actionable button/option attached to a NotificationPayload.

id class-attribute instance-attribute

id: str = Field(description='Stable identifier for this action, used when the user selects it.')

label class-attribute instance-attribute

label: str = Field(description='Human-readable label shown to the user.')

payload class-attribute instance-attribute

payload: dict[str, Any] = Field(default_factory=dict, description='Arbitrary data returned to the caller when this action is selected.')

planner

Planner task pipeline for streaming sessions.

PlannerTaskPipeline

PlannerTaskPipeline(*, planner_factory: PlannerFactory, event_sink: PlannerEventSink | None = None)

Run a ReactPlanner inside a StreamingSession task.