Skip to content
Pyrula

Workflows

Generated from the type stubs and docstrings. Do not edit by hand.

Types

Section titled “Types”

BaseContext

Section titled “BaseContext”

Base durable execution context with step/sleep/signal/timers/determinism.

Shared by all run-kinds. Extended by run-kind-specific subclasses that add their own capabilities (e.g. LLM streaming, tool loops).

Fields:

  • run_id: str
  • metadata: dict[str, Any]
  • deps: Optional[DepsT] = None
  • principal: Any = None
  • state: Optional[RunState] = None
  • cancelled: bool

checkpoint

Section titled “checkpoint”
checkpoint(self) -> None

step

Section titled “step”
step(self, step_id: str, fn: Any, *args: Any, step_kind: str = StepKind.NAMED, effect: str = 'write', **kwargs: Any) -> Any

Execute fn as a durable named step. Returns stored result on replay without re-executing.

step_id must be stable across retries. Results must be JSON-serializable. Sync callables are run in a thread. Async callables are awaited directly. step_kind should be StepKind.NAMED (default) or StepKind.DATA for data steps. effect declares the step’s side-effect class: “read” | “write” (default) | “pure”. The cloud uses this to set smart per-step replay defaults (read/pure → live, write → captured).

now

Section titled “now”
now(self) -> float

Replay-stable Unix timestamp (seconds). Use instead of time.time() inside run bodies.

uuid

Section titled “uuid”
uuid(self) -> str

Replay-stable UUID hex string. Use instead of uuid.uuid4().hex inside run bodies.

random

Section titled “random”
random(self) -> float

Replay-stable float in [0.0, 1.0). Use instead of random.random() inside run bodies.

emit

Section titled “emit”
emit(self, kind: str, payload: dict[str, Any], id: Optional[str] = None) -> RunEvent

emit_status

Section titled “emit_status”
emit_status(self, text: str) -> RunEvent

sleep

Section titled “sleep”
sleep(self, duration: timedelta | float) -> None

Timer re-queues after wake_time. Worker slot freed.

sleep_until

Section titled “sleep_until”
sleep_until(self, dt: datetime) -> None

Returns immediately if dt is already past.

wait_for_signal

Section titled “wait_for_signal”
wait_for_signal(self, name: str) -> Signal

interrupt

Section titled “interrupt”
interrupt(self, event_id: str, payload: dict[str, Any]) -> InterruptResponse

stream

Section titled “stream”
stream(self, source: Source[Any], timeout_ms: int = 5000) -> AsyncIterator[Batch[Any]]

Yield batches from source, replay-skipping already-committed offsets.

On resume, seeks past the last committed watermark for each partition before the first poll. Each batch is yielded inside a scope (source_id:partition:offset) so that nested primitives (step, checkpoint, sleep, …) are keyed to that batch and replayed in isolation.

commit

Section titled “commit”
commit(self, source: Source[Any], batch: Batch[Any]) -> None

Persist a batch’s offset for exactly-once semantics.

Must be called after each successful batch processing step. On resume, ctx.stream() seeks past this offset.

Batch

Section titled “Batch”

A batch of records from a source.

Fields:

  • source_id: str
  • partition: int
  • offset: int
  • records: list[T]

CacheStore

Section titled “CacheStore”

cache_get

Section titled “cache_get”
cache_get(self, key: str) -> Optional[Any]

cache_set

Section titled “cache_set”
cache_set(self, key: str, value: Any, ttl: float) -> None

CompositeStore

Section titled “CompositeStore”

Combines a lifecycle backend with an event-log backend.

Usage::

store = CompositeStore(
    lifecycle=ValkeyLifecycleStore("redis://..."),
    events=DeltaEventLogStore("s3://lake/runs"),
)

Lifecycle operations delegate to lifecycle. Event-log operations delegate to events.

NOTE: the optional-mixin operations (cache, idempotency, threads, timers, KV, signals) also delegate to lifecycle. The parameter is typed LifecycleStore for ergonomics, but a backend used with the agents engine must also provide whichever mixin methods the runtime exercises - a strictly minimal LifecycleStore will raise AttributeError when those paths are hit.

submit_run

Section titled “submit_run”
submit_run(self, name: str, run_id: str, params: dict[str, object], **kw: Any) -> str

claim_next_run

Section titled “claim_next_run”
claim_next_run(self, name: str, worker_id: str) -> Optional[str]

release_run

Section titled “release_run”
release_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None

requeue_run

Section titled “requeue_run”
requeue_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None

recover_pending_runs

Section titled “recover_pending_runs”
recover_pending_runs(self, name: str, now: Optional[float] = None) -> list[str]

get_pending_count

Section titled “get_pending_count”
get_pending_count(self, name: str) -> int

heartbeat

Section titled “heartbeat”
heartbeat(self, name: str, run_id: str, worker_id: str) -> None

request_cancel

Section titled “request_cancel”
request_cancel(self, name: str, run_id: str) -> str

get_cancel_requested_at

Section titled “get_cancel_requested_at”
get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str]

expire_run

Section titled “expire_run”
expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool

append_event

Section titled “append_event”
append_event(self, name: str, run_id: str, event: RunEvent) -> None

replay_run

Section titled “replay_run”
replay_run(self, name: str, run_id: str, **kw: Any) -> tuple[list[RunEvent], Optional[Any]]

load_run

Section titled “load_run”
load_run(self, name: str, run_id: str, **kw: Any) -> list[RunEvent]

get_last_event

Section titled “get_last_event”
get_last_event(self, name: str, run_id: str) -> Optional[RunEvent]

read_entries_from

Section titled “read_entries_from”
read_entries_from(self, name: str, run_id: str, cursor: Optional[str], count: int) -> tuple[list[tuple[str, RunEvent]], Optional[str]]

set_run_owner

Section titled “set_run_owner”
set_run_owner(self, name: str, run_id: str, owner: str) -> None

get_run_owner

Section titled “get_run_owner”
get_run_owner(self, name: str, run_id: str) -> Optional[str]

ping

Section titled “ping”
ping(self) -> None

list_runs

Section titled “list_runs”
list_runs(self, name: str, limit: int = 50, cursor: Optional[str] = None) -> tuple[list[dict[str, Any]], Optional[str]]

set_interrupted

Section titled “set_interrupted”
set_interrupted(self, name: str, run_id: str, event_id: str, payload: dict[str, Any]) -> None

resume_run

Section titled “resume_run”
resume_run(self, name: str, run_id: str, action: str, value: Any) -> None

complete_run

Section titled “complete_run”
complete_run(self, name: str, run_id: str) -> None

get_interrupt_response

Section titled “get_interrupt_response”
get_interrupt_response(self, name: str, run_id: str) -> Optional[dict[str, Any]]

get_idempotency_run

Section titled “get_idempotency_run”
get_idempotency_run(self, name: str, key: str) -> Optional[str]

set_idempotency_run

Section titled “set_idempotency_run”
set_idempotency_run(self, name: str, key: str, run_id: str, ttl: int = 86400) -> bool

cache_get

Section titled “cache_get”
cache_get(self, key: str) -> Optional[Any]

cache_set

Section titled “cache_set”
cache_set(self, key: str, value: Any, ttl: float) -> None

create_thread

Section titled “create_thread”
create_thread(self, thread_id: str, name: str, **kw: Any) -> None

get_thread

Section titled “get_thread”
get_thread(self, thread_id: str) -> Optional[StoreItem]

delete_thread

Section titled “delete_thread”
delete_thread(self, thread_id: str) -> None

add_run_to_thread

Section titled “add_run_to_thread”
add_run_to_thread(self, thread_id: str, name: str, run_id: str) -> None

set_thread_completed

Section titled “set_thread_completed”
set_thread_completed(self, thread_id: str, run_id: str) -> None

get_thread_runs

Section titled “get_thread_runs”
get_thread_runs(self, thread_id: str, limit: int = 50) -> list[str]

get_thread_last_result

Section titled “get_thread_last_result”
get_thread_last_result(self, thread_id: str) -> Optional[Any]

get_thread_history

Section titled “get_thread_history”
get_thread_history(self, thread_id: str, limit: int = 50) -> list[StoreItem]

set_sleeping

Section titled “set_sleeping”
set_sleeping(self, name: str, run_id: str, wake_time: float) -> None

get_sleeping_runs

Section titled “get_sleeping_runs”
get_sleeping_runs(self, before: float, limit: int = 100) -> list[tuple[str, str]]

wake_run

Section titled “wake_run”
wake_run(self, name: str, run_id: str) -> None

kv_get

Section titled “kv_get”
kv_get(self, namespace: str, key: str) -> Optional[dict[str, Any]]

kv_put

Section titled “kv_put”
kv_put(self, namespace: str, key: str, value: dict[str, Any]) -> None

kv_delete

Section titled “kv_delete”
kv_delete(self, namespace: str, key: str) -> None

kv_list

Section titled “kv_list”
kv_list(self, namespace: str, prefix: str = '', limit: int = 100) -> list[dict[str, Any]]
Section titled “kv_search”
kv_search(self, namespace: str, query: str, limit: int = 10) -> list[dict[str, Any]]

send_signal

Section titled “send_signal”
send_signal(self, name: str, run_id: str, signal_name: str, payload: dict[str, Any], sent_at: str) -> str

get_buffered_signal

Section titled “get_buffered_signal”
get_buffered_signal(self, name: str, run_id: str, signal_name: str, call_index: int) -> Optional[dict[str, Any]]

set_signal_waiting

Section titled “set_signal_waiting”
set_signal_waiting(self, name: str, run_id: str, signal_name: str, call_index: int) -> None

arm_signal_wait

Section titled “arm_signal_wait”
arm_signal_wait(self, name: str, run_id: str, signal_name: str, call_index: int) -> Optional[dict[str, Any]]

EventKind

Section titled “EventKind”

Core event kinds - lifecycle + step only.

Agent-specific kinds (LLM_, BLOCK_, TOOL_, AGENT_CALL_) live in pyrula.agents’s EventKind and ride the wire as plain strings.

Fields:

  • RUN_INIT = 'run:init'
  • RUN_CANCEL = 'run:cancel'
  • RUN_PAUSE_REQUESTED = 'run:pause_requested'
  • RUN_PAUSED = 'run:paused'
  • RUN_REQUEUED = 'run:requeued'
  • RUN_STATUS = 'run:status'
  • RUN_COMPLETE = 'run:complete'
  • RUN_ERROR = 'run:error'
  • RUN_INTERRUPTED = 'run:interrupted'
  • RUN_RESUMED = 'run:resumed'
  • RUN_SLEEPING = 'run:sleeping'
  • RUN_WOKEN = 'run:woken'
  • RUN_SIGNAL_WAITING = 'run:signal_waiting'
  • RUN_SIGNAL_RECEIVED = 'run:signal_received'
  • STEP_PENDING = 'step:pending'
  • STEP_DONE = 'step:done'
  • STEP_ERROR = 'step:error'
  • HEARTBEAT = 'heartbeat'
  • STREAM_OVERFLOW = 'stream:overflow'

EventLogStore

Section titled “EventLogStore”

Required for event storage: append ordered events, read ranges, load/replay a run.

Backends that only implement this (e.g. Kafka, Delta) provide event transport/archival but not claim/lease coordination. Pair with a LifecycleStore via CompositeStore for full durability.

append_event

Section titled “append_event”
append_event(self, name: str, run_id: str, event: RunEvent) -> None

load_run

Section titled “load_run”
load_run(self, name: str, run_id: str, max_events: Optional[int] = None) -> list[RunEvent]

replay_run

Section titled “replay_run”
replay_run(self, name: str, run_id: str, cursor: Optional[Any] = None, count: Optional[int] = None) -> tuple[list[RunEvent], Optional[Any]]

get_last_event

Section titled “get_last_event”
get_last_event(self, name: str, run_id: str) -> Optional[RunEvent]

read_entries_from

Section titled “read_entries_from”
read_entries_from(self, name: str, run_id: str, cursor: Optional[str], count: int) -> tuple[list[tuple[str, RunEvent]], Optional[str]]

IdempotencyStore

Section titled “IdempotencyStore”

get_idempotency_run

Section titled “get_idempotency_run”
get_idempotency_run(self, name: str, key: str) -> Optional[str]

set_idempotency_run

Section titled “set_idempotency_run”
set_idempotency_run(self, name: str, key: str, run_id: str, ttl: int = 86400) -> bool

InMemorySource

Section titled “InMemorySource”

In-memory source for testing pipeline streaming.

Records are provided at construction. poll() returns batches one at a time, advancing the internal cursor. Committed offsets are tracked in-memory.

Fields:

  • source_id: str

poll

Section titled “poll”
poll(self, timeout_ms: int = 5000) -> Optional[Batch[T]]

seek

Section titled “seek”
seek(self, source_id: str, partition: int, offset: int) -> None

get_committed

Section titled “get_committed”
get_committed(self, source_id: str, partition: int) -> Optional[int]

partitions

Section titled “partitions”
partitions(self) -> list[tuple[str, int]]

inject_committed

Section titled “inject_committed”
inject_committed(self, source_id: str, partition: int, offset: int) -> None

Test helper: set committed offset directly.

InlineWorkflowRunner

Section titled “InlineWorkflowRunner”

Single-process workflow runner for testing and embedded use.

When cloud=True (default) and PYRULA_CLOUD_API_KEY is set, events are automatically ingested to Pyrula Cloud.

submit

Section titled “submit”
submit(self, name: str, params: Optional[dict[str, object]] = None, metadata: Optional[dict[str, Any]] = None) -> str

run_next

Section titled “run_next”
run_next(self, name: str) -> Optional[list[RunEvent]]

run

Section titled “run”
run(self, name: str, run_id: str) -> Optional[list[RunEvent]]

aclose

Section titled “aclose”
aclose(self) -> None

InterruptAction

Section titled “InterruptAction”

Fields:

  • ACCEPT = 'accept'
  • EDIT = 'edit'
  • RESPOND = 'respond'
  • IGNORE = 'ignore'

InterruptResponse

Section titled “InterruptResponse”

Fields:

  • action: InterruptAction | str
  • value: Any = None

KVStore

Section titled “KVStore”

kv_get

Section titled “kv_get”
kv_get(self, namespace: str, key: str) -> Optional[dict[str, Any]]

kv_put

Section titled “kv_put”
kv_put(self, namespace: str, key: str, value: dict[str, Any]) -> None

kv_delete

Section titled “kv_delete”
kv_delete(self, namespace: str, key: str) -> None

kv_list

Section titled “kv_list”
kv_list(self, namespace: str, prefix: str = '', limit: int = 100) -> list[dict[str, Any]]

kv_search

Section titled “kv_search”
kv_search(self, namespace: str, query: str, limit: int = 10) -> list[dict[str, Any]]

LifecycleStore

Section titled “LifecycleStore”

Required: run queue, ownership, cancellation, heartbeat, run listing.

This is the minimal interface for claim/lease coordination. Event storage is EventLogStore. Backends that provide both (MemoryStore, ValkeyStore, PostgresStore) implement Store.

submit_run

Section titled “submit_run”
submit_run(self, name: str, run_id: str, params: dict[str, object], metadata: Optional[dict[str, Any]] = None, owner: Optional[str] = None, idempotency_key: Optional[str] = None) -> str

claim_next_run

Section titled “claim_next_run”
claim_next_run(self, name: str, worker_id: str) -> Optional[str]

release_run

Section titled “release_run”
release_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None

requeue_run

Section titled “requeue_run”
requeue_run(self, name: str, run_id: str, worker_id: Optional[str] = None) -> None

recover_pending_runs

Section titled “recover_pending_runs”
recover_pending_runs(self, name: str, now: Optional[float] = None) -> list[str]

get_pending_count

Section titled “get_pending_count”
get_pending_count(self, name: str) -> int

heartbeat

Section titled “heartbeat”
heartbeat(self, name: str, run_id: str, worker_id: str) -> None

request_cancel

Section titled “request_cancel”
request_cancel(self, name: str, run_id: str) -> str

get_cancel_requested_at

Section titled “get_cancel_requested_at”
get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str]

expire_run

Section titled “expire_run”
expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool

set_run_owner

Section titled “set_run_owner”
set_run_owner(self, name: str, run_id: str, owner: str) -> None

get_run_owner

Section titled “get_run_owner”
get_run_owner(self, name: str, run_id: str) -> Optional[str]

ping

Section titled “ping”
ping(self) -> None

list_runs

Section titled “list_runs”
list_runs(self, name: str, limit: int = 50, cursor: Optional[str] = None) -> tuple[list[dict[str, Any]], Optional[str]]

set_interrupted

Section titled “set_interrupted”
set_interrupted(self, name: str, run_id: str, event_id: str, payload: dict[str, Any]) -> None

resume_run

Section titled “resume_run”
resume_run(self, name: str, run_id: str, action: str, value: Any) -> None

complete_run

Section titled “complete_run”
complete_run(self, name: str, run_id: str) -> None

get_interrupt_response

Section titled “get_interrupt_response”
get_interrupt_response(self, name: str, run_id: str) -> Optional[dict[str, Any]]

MemoryStore

Section titled “MemoryStore”

In-memory store implementation for testing only.

This store does not persist data across process restarts. Use ValkeyStore or PostgresStore for production deployments.

Fields:

  • heartbeat_interval = heartbeat_interval
  • lock_ttl = lock_ttl
  • max_replay_entries = max_replay_entries
  • max_retries = max_retries
  • max_thread_runs = max_thread_runs

load_run

Section titled “load_run”
load_run(self, name: str, run_id: str, max_events: Optional[int] = None) -> list[RunEvent]

expire_run

Section titled “expire_run”
expire_run(self, name: str, run_id: str, ttl_seconds: int = 360) -> bool

ReplayState

Section titled “ReplayState”

Core replay state for durable run reconstruction.

Profile-specific fields (e.g. LLM messages, tool results) live in profile-specific subclasses (e.g. pyrula.agents’s AgentReplayState).

Fields:

  • recorded_call_results: dict[tuple[str, int], Any] = field(default_factory=dict)
  • recorded_call_results_by_id: dict[tuple[str, str], Any] = field(default_factory=dict)
  • manual_checkpoint_counts: dict[str, int] = field(default_factory=dict)
  • emit_counts: dict[str, int] = field(default_factory=dict)
  • fanout_results: dict[str, list[Any]] = field(default_factory=dict)
  • fanout_pending: dict[str, list[dict[str, str]]] = field(default_factory=dict)
  • step_done_worker: Optional[str] = None
  • deterministic_results: dict[tuple[str, int], Any] = field(default_factory=dict)
  • interrupt_results: dict[str, dict[str, Any]] = field(default_factory=dict)
  • sleep_results: dict[str, set[int]] = field(default_factory=dict)
  • signal_results: dict[tuple[str, int], dict[str, Any]] = field(default_factory=dict)
  • named_step_results: dict[tuple[str, int], Any] = field(default_factory=dict)
  • captured_step_keys: set[tuple[str, int]] = field(default_factory=set)
  • stream_committed_offsets: dict[tuple[str, int], int] = field(default_factory=dict)

RunEvent

Section titled “RunEvent”

A run event - persisted in the event log and returned by replay.

schema_version is the wire-format version. When a runtime reads an event whose schema_version is higher than what it understands, replay MUST reject with a helpful error rather than silently misinterpreting the payload.

Fields:

  • kind: str
  • payload: dict[str, Any]
  • run_id: str
  • schema_version: str = EVENT_SCHEMA_VERSION
  • run_seq: Optional[int] = None

RunExecutor

Section titled “RunExecutor”

Executes ONE attempt of ONE already-claimed run. Does not claim/release.

prime_replay_policy

Section titled “prime_replay_policy”
prime_replay_policy(self, run_id: str, policy: tuple[list[RunEvent], int, dict[str, str]]) -> None

Stash a Model C replay policy consumed once by _build_state for run_id.

run_attempt

Section titled “run_attempt”
run_attempt(self, name: str, run_id: str) -> list[RunEvent]

Run one attempt. Returns the public event list for the run.

RunRef

Section titled “RunRef”

Fields:

  • name: str
  • run_id: str

RunReporter

Section titled “RunReporter”

Optional sink for run lifecycle/events. Default impl is NoopReporter.

A reporter that wants live per-event publication returns True from wants_live_sequencing() and supplies a persisted listener; the executor then wires the store’s sequenced writer so each durable append is published with its canonical run_seq.

on_event

Section titled “on_event”
on_event(self, name: str, run_id: str, event: RunEvent, seq: int) -> None

on_attempt_complete

Section titled “on_attempt_complete”
on_attempt_complete(self, name: str, run_id: str, state: Any, new_events: list[RunEvent], seq_base: int, existing_seqs: set[int], is_first_run: bool) -> None

wants_live_sequencing

Section titled “wants_live_sequencing”
wants_live_sequencing(self) -> bool

RunState

Section titled “RunState”

Fields:

  • run_id: str
  • events: list[RunEvent]
  • name: str = ''
  • deps: Any = None
  • public_events: list[RunEvent] = field(default_factory=list)
  • event_writer: Optional[Callable[[RunEvent], None]] = None
  • sequenced_event_writer: Optional[Callable[[RunEvent], int]] = None
  • persisted_event_listener: Optional[PersistedEventListener] = None
  • stream_writer: Optional[Callable[[RunEvent], None]] = None
  • worker_id: Optional[str] = None
  • replay_state: Optional[ReplayState] = None
  • cancellation_checker: Optional[Callable[[], Optional[str]]] = None
  • pause_checker: Optional[Callable[[], Optional[str]]] = None
  • manual_checkpoint_counts: dict[str, int] = field(default_factory=dict)
  • emit_counts: dict[str, int] = field(default_factory=dict)
  • recorded_call_counts: dict[str, int] = field(default_factory=dict)
  • recorded_call_ids: set[tuple[str, str]] = field(default_factory=set)
  • cancelled_at: Optional[str] = None
  • paused_at: Optional[str] = None
  • cache_getter: Optional[Callable[[str], Optional[Any]]] = None
  • cache_setter: Optional[Callable[[str, Any, float], None]] = None
  • store: Optional[Any] = field(default=None)
  • thread_id: Optional[str] = None
  • step_done_worker: Optional[str] = None
  • scope_stack: list[str] = field(default_factory=list)
  • deterministic_call_counts: dict[str, int] = field(default_factory=dict)
  • sleep_call_counts: dict[str, int] = field(default_factory=dict)
  • signal_call_counts: dict[str, int] = field(default_factory=dict)
  • named_step_call_counts: dict[str, int] = field(default_factory=dict)

scoped

Section titled “scoped”
scoped(self, base: str) -> str

Build a scope-keyed replay base. Empty stack returns base unchanged.

write

Section titled “write”
write(self, kind: str, payload: Optional[dict[str, Any]] = None, public: bool = True, durable: bool = True) -> RunEvent

write_async

Section titled “write_async”
write_async(self, kind: str, payload: Optional[dict[str, Any]] = None, public: bool = True, durable: bool = True) -> RunEvent

poll_cancellation

Section titled “poll_cancellation”
poll_cancellation(self) -> Optional[str]

poll_pause

Section titled “poll_pause”
poll_pause(self) -> Optional[str]

Mirror of poll_cancellation for cooperative pause. Returns the request timestamp once a pause has been requested for this run, else None. The caller decides where to honor it (a clean step boundary), so unlike cancellation this is read at suspend-safe points, not mid-stream.

RunStatus

Section titled “RunStatus”

Fields:

  • PENDING = 'pending'
  • IN_PROGRESS = 'in_progress'
  • COMPLETE = 'complete'
  • CANCELLED = 'cancelled'
  • ERROR = 'error'
  • QUARANTINED = 'quarantined'
  • INTERRUPTED = 'interrupted'
  • PAUSED = 'paused'
  • SLEEPING = 'sleeping'
  • SIGNAL_WAITING = 'signal_waiting'

SchemaVersionMismatch

Section titled “SchemaVersionMismatch”

Raised when an event’s schema_version is newer than the runtime understands.

Signal

Section titled “Signal”

Fields:

  • name: str
  • payload: dict[str, Any]
  • sent_at: str

SignalStore

Section titled “SignalStore”

send_signal

Section titled “send_signal”
send_signal(self, name: str, run_id: str, signal_name: str, payload: dict[str, Any], sent_at: str) -> str

get_buffered_signal

Section titled “get_buffered_signal”
get_buffered_signal(self, name: str, run_id: str, signal_name: str, call_index: int) -> Optional[dict[str, Any]]

set_signal_waiting

Section titled “set_signal_waiting”
set_signal_waiting(self, name: str, run_id: str, signal_name: str, call_index: int) -> None

arm_signal_wait

Section titled “arm_signal_wait”
arm_signal_wait(self, name: str, run_id: str, signal_name: str, call_index: int) -> Optional[dict[str, Any]]

Atomically take a buffered signal if queued, else arm the waiting marker. Returns the buffered signal dict (deliver inline) or None (suspend). One op so a concurrent send_signal cannot race the buffer-check/arm and lose the signal.

Source

Section titled “Source”

Protocol for stream sources.

Implementations yield Batch objects and track committed offsets. The engine calls seek() on resume and commit() after a successful step.

Fields:

  • source_id: str Unique identifier for this source (e.g. topic name).

poll

Section titled “poll”
poll(self, timeout_ms: int = 5000) -> Optional[Batch[T]]

Return the next batch, or None if no data within timeout.

seek

Section titled “seek”
seek(self, source_id: str, partition: int, offset: int) -> None

Seek to a specific offset (used on resume).

get_committed

Section titled “get_committed”
get_committed(self, source_id: str, partition: int) -> Optional[int]

Return the last committed offset for a partition, or None.

partitions

Section titled “partitions”
partitions(self) -> list[tuple[str, int]]

Return all (source_id, partition) pairs.

StepKind

Section titled “StepKind”

Fields:

  • MANUAL = 'manual'
  • EMIT = 'emit'
  • FANOUT = 'fanout'
  • DETERMINISTIC = 'deterministic'
  • INTERRUPT = 'interrupt'
  • SLEEP = 'sleep'
  • SIGNAL = 'signal'
  • NAMED = 'named'
  • DATA = 'data'
  • STREAM = 'stream'

Store

Section titled “Store”

Full store interface: lifecycle + event-log + all optional mixins.

Existing backends (MemoryStore, ValkeyStore, PostgresStore) implement this complete interface. Custom backends should implement LifecycleStore and/or EventLogStore plus only the optional mixins they need.

StoreItem

Section titled “StoreItem”

Fields:

  • key: str
  • value: dict[str, Any]
  • updated_at: float

ThreadStore

Section titled “ThreadStore”

create_thread

Section titled “create_thread”
create_thread(self, thread_id: str, name: str, metadata: Optional[dict[str, Any]] = None, owner: Optional[str] = None) -> None

get_thread

Section titled “get_thread”
get_thread(self, thread_id: str) -> Optional[StoreItem]

delete_thread

Section titled “delete_thread”
delete_thread(self, thread_id: str) -> None

add_run_to_thread

Section titled “add_run_to_thread”
add_run_to_thread(self, thread_id: str, name: str, run_id: str) -> None

set_thread_completed

Section titled “set_thread_completed”
set_thread_completed(self, thread_id: str, run_id: str) -> None

get_thread_runs

Section titled “get_thread_runs”
get_thread_runs(self, thread_id: str, limit: int = 50) -> list[str]

get_thread_last_result

Section titled “get_thread_last_result”
get_thread_last_result(self, thread_id: str) -> Optional[Any]

get_thread_history

Section titled “get_thread_history”
get_thread_history(self, thread_id: str, limit: int = 50) -> list[StoreItem]

TimerStore

Section titled “TimerStore”

set_sleeping

Section titled “set_sleeping”
set_sleeping(self, name: str, run_id: str, wake_time: float) -> None

get_sleeping_runs

Section titled “get_sleeping_runs”
get_sleeping_runs(self, before: float, limit: int = 100) -> list[tuple[str, str]]

wake_run

Section titled “wake_run”
wake_run(self, name: str, run_id: str) -> None

UnboundContextError

Section titled “UnboundContextError”

WorkerOptions

Section titled “WorkerOptions”

Tuning knobs for WorkflowWorker.

Fields:

  • max_concurrent_runs: int = 50
  • claim_poll_interval: float = 0.1
  • timer_poll_interval_s: float = 5.0
  • timer_max_batch: int = 100
  • orphan_scan_interval: float = 60.0
  • shutdown_grace: float = 30.0
  • cancel_timeout: float = 5.0
  • poison_max_retries: int = 3
  • poison_retry_ttl_s: float = 3600.0
  • reconcile_interval: float = 60.0

WorkflowProfile

Section titled “WorkflowProfile”

Run-kind profile for workflows (no LLM).

build_context

Section titled “build_context”
build_context(self, run_id: str, state: RunState, metadata: dict[str, Any], deps: Any = None) -> BaseContext

run_body

Section titled “run_body”
run_body(self, fn: Callable[..., Any], ctx: BaseContext, params: dict[str, Any], spec: Any, timeout: float) -> Any

finalize

Section titled “finalize”
finalize(self, state: RunState, result: Any, name: str, run_id: str, start_time: Optional[float] = None) -> None

WorkflowSpec

Section titled “WorkflowSpec”

Fields:

  • name: str
  • timeout: float = 300.0
  • max_event_count_per_run: Optional[int] = None
  • max_event_payload_bytes: Optional[int] = None
  • max_run_wall_seconds: Optional[float] = None

WorkflowWorker

Section titled “WorkflowWorker”

Run-vocab supervisor: per-workflow claim loops + timer + orphan + shutdown.

Drives a RunExecutor against a shared Store so durable workflows run in production without the agent runtime. Reporting (e.g. cloud) is optional via the reporter seam.

Shares WorkerSupervisorBase (stop event, concurrency slots, poll/sleep helpers) with the agents WorkerRuntime; the two are siblings - the agents worker speaks turn-vocab and owns its own LLM/poison machinery rather than subclassing this one.

submit

Section titled “submit”
submit(self, name: str, params: Optional[dict[str, Any]] = None, metadata: Optional[dict[str, Any]] = None) -> str

Submit a new run and return its run_id.

run

Section titled “run”
run(self) -> None

Run all supervisor loops until stop() is called.

Functions

Section titled “Functions”

decode_result

Section titled “decode_result”
decode_result(encoded: dict[str, Any]) -> Any

Reverse encode_result, unwrapping an envelope back to a Python value.

Custom pyrula types (IList, WriteReport, …) come back as plain lists/dicts since their classes may not be importable when replaying.

detect_version_skew

Section titled “detect_version_skew”
detect_version_skew(state: RunState) -> Optional[dict[str, Any]]

encode_result

Section titled “encode_result”
encode_result(obj: Any) -> dict[str, Any]

Wrap a step result in a JSON-safe {“type”, “value”} envelope.

reconstruct_state

Section titled “reconstruct_state”
reconstruct_state(events: list[RunEvent], state: Optional[ReplayState] = None, extra_arms: Optional[dict[str, ArmHandler]] = None) -> ReplayState

Reconstruct replay state from events.

Core neutral arms are single-source. Profiles (pyrula.agents) register agent-specific arms via extra_arms - never by copying neutral code.

Raises SchemaVersionMismatch if any event carries a schema_version newer than the runtime’s EVENT_SCHEMA_VERSION.

replay_all_events

Section titled “replay_all_events”
replay_all_events(store: Store, name: str, run_id: str) -> list[RunEvent]