Workflows

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

Types

`ArchivingStore`

Composes a live Store + a durable Archive.

Live-first reads with archive fallback; on terminal append, defer live retention, archive the completed stream, then apply retention. Crash-safe via a KV pending set kept in the live tier (consistent with the durability boundary: completed-but-unarchived data lives only in the live tier until archived, so the marker lives there too).

`append_event`

append_event(self, name: str, run_id: str, event: RunEvent) -> None

`append_event_sequenced`

append_event_sequenced(self, name: str, run_id: str, event: RunEvent, retain: bool = True) -> int

`append_events_sequenced`

append_events_sequenced(self, name: str, run_id: str, events: list[RunEvent], retain: bool = True) -> list[int]

`append_events`

append_events(self, name: str, run_id: str, events: list[RunEvent]) -> None

`sync_run`

sync_run(self, name: str, run_id: str) -> int

Archive whatever is currently in the live stream, no terminal event required - an operator repair for a missed archive. Idempotent (archive_run upserts). Returns the number of events archived. Does NOT apply retention (the run may still be live).

`reconcile_pending_archives`

reconcile_pending_archives(self) -> int

Finish (or retry) archival for runs left pending by a crash or archive outage. Idempotent (archive_run upserts; apply_retention repeats safely). Returns the number of runs reconciled.

`register_projector`

register_projector(self, projector: Projector) -> None

`reproject_run`

reproject_run(self, name: str, run_id: str) -> None

Re-run the registered projectors for one archived run (backfill / rebuild). Idempotent - projectors upsert.

`reproject_all`

reproject_all(self, name: Optional[str] = None) -> int

Re-project archived runs. With name given, only that workflow’s runs; otherwise every archived run. Returns the number of archived runs visited (each run’s registered projectors are re-run; with no projector registered the visit is a no-op).

`drain_projection_pending`

drain_projection_pending(self) -> int

Project each pending completed run from the archive (SoR). Idempotent (projectors upsert). A failure leaves the run pending for the next sweep, same backstop as reconcile_pending_archives. Returns runs drained.

`load_run`

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

`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`

get_last_event(self, name: str, run_id: str) -> Optional[RunEvent]

`read_entries_from`

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

`get_run_owner`

get_run_owner(self, name: str, run_id: str) -> Optional[str]

`list_names`

list_names(self) -> list[str]

`apply_retention`

apply_retention(self, name: str, run_id: str) -> None

`arm_signal_wait`

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

`cache_get`

cache_get(self, key: str) -> Optional[Any]

`cache_set`

cache_set(self, key: str, value: Any, ttl: float) -> None

`claim_next_run`

claim_next_run(self, name: str, worker_id: str, worker_version: Optional[str] = None) -> Optional[str]

`claim_run`

claim_run(self, name: str, run_id: str, worker_id: str) -> None

`complete_run`

complete_run(self, name: str, run_id: str) -> None

`deliver_update`

deliver_update(self, name: str, run_id: str, update_name: str, payload: dict[str, Any], update_id: str, sent_at: str) -> str

`drain_pending_updates`

drain_pending_updates(self, name: str, run_id: str) -> list[Any]

`expire_run`

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

`get_buffered_signal`

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

`get_cancel_requested_at`

get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str]

`get_idempotency_run`

get_idempotency_run(self, name: str, key: str) -> Optional[str]

`get_interrupt_response`

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

`get_pending_count`

get_pending_count(self, name: str) -> int

`get_sleeping_runs`

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

`heartbeat`

heartbeat(self, name: str, run_id: str, worker_id: str) -> None

`kv_delete`

kv_delete(self, namespace: str, key: str) -> None

`kv_get`

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

`kv_list`

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

`kv_put`

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

`kv_search`

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

`list_runs`

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

`ping`

ping(self) -> None

`read_update_result`

read_update_result(self, name: str, run_id: str, update_id: str) -> Optional[dict[str, Any]]

`recover_pending_runs`

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

`release_run`

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

`request_cancel`

request_cancel(self, name: str, run_id: str) -> str

`requeue_run`

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

`resume_run`

resume_run(self, name: str, run_id: str, action: str, value: Any) -> None

`send_signal`

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

`set_idempotency_run`

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

`set_interrupted`

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

`set_run_owner`

set_run_owner(self, name: str, run_id: str, owner: str) -> None

`set_signal_waiting`

set_signal_waiting(self, name: str, run_id: str, signal_name: str, call_index: int) -> None

`set_sleeping`

set_sleeping(self, name: str, run_id: str, wake_time: float) -> None

`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, idempotency_body_hash: Optional[str] = None, idempotency_ttl: int = 86400, version: str = '', version_behavior: str = 'pinned', concurrency_key: Optional[str] = None, concurrency_max: Optional[int] = None, rate_key: Optional[str] = None, rate_per_second: Optional[float] = None, rate_burst: Optional[int] = None) -> str

`wake_run`

wake_run(self, name: str, run_id: str) -> None

`write_update_result`

write_update_result(self, name: str, run_id: str, update_id: str, outcome: dict[str, Any]) -> None

`adopt_run`

adopt_run(self, name: str, run_id: str, new_version: str) -> None

`get_pause_requested_at`

get_pause_requested_at(self, name: str, run_id: str) -> Optional[str]

`get_run_version`

get_run_version(self, name: str, run_id: str) -> Optional[tuple[str, str]]

`submitted_at`

submitted_at(self, name: str, run_id: str) -> Optional[float]

`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 Whether cancellation has been requested for this run.

Cancellation is cooperative/poll-based, not preemptive: it does NOT interrupt an in-flight ctx.step / ctx.gather / ctx.sleep. Those primitives run to their natural boundary; only ctx.run_child’s wait loop and code that explicitly reads ctx.cancelled observe it. Check ctx.cancelled at your own long-running loop boundaries to bail out early (Temporal-style cancellation-is-a-request semantics).

continue_suggested: bool True once this run’s event log has consumed >= 80% of its max_event_count_per_run cap (Temporal’s suggest-continue idiom): call ctx.continue_as_new at a convenient boundary when this turns True. Always False when the run is uncapped. Pure property, no I/O.

Caveat for kafka-driven turns: ctx.continue_as_new is rejected there (raises ContinueAsNewNotSupportedError) - the consumed record/batch is the work unit, so there is no chain owner to hand a continuation to. A kafka handler that observes this turn True cannot continue_as_new; split the handler into smaller steps, or raise the event-count cap instead.

`checkpoint`

checkpoint(self) -> None

`step`

step(self, step_id: str, fn: Any, *args: Any, step_kind: str = StepKind.NAMED, effect: str = 'write', retry: Optional[RetryPolicy] = None, timeout: Optional[float] = None, **kwargs: Any) -> Any

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

A step is where non-determinism belongs. Call time.time(), uuid4(), an HTTP API, a database - the result is journaled and replay returns the recorded value instead of re-running fn, so the step boundary IS the determinism mechanism. The rule to hold in your head is:

inside a step, anything goes; outside one, the workflow body must be
deterministic - use ctx.now / ctx.uuid / ctx.random there.

Prefer ctx.sleep over time.sleep in a step: only ctx.sleep survives a crash, and a blocking sleep holds a worker thread for its duration.

step_id must be stable across retries. 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.

The result is stored as JSON, so it comes back as its JSON projection - live and on replay alike, since the value is normalized before it is recorded. Two coercions are easy to miss because they are silent:

(1, 2)        -> [1, 2]            tuples become lists
{1: "a"}      -> {"1": "a"}        non-string dict keys become strings

Models and dataclasses become dicts; datetime/date/time become ISO-8601 strings, Decimal and UUID strings, set a list. A value with no JSON projection (bytes, timedelta, a socket) fails the step rather than being recorded as a repr that could not round-trip, and so do NaN/Infinity, which are not JSON. See pyrula.contracts._jsonable.

effect declares the step’s side-effect class: “read” | “write” (default) | “pure”. The cloud uses it to set per-step replay defaults on a counterfactual branch (read/pure -> re-run live, write -> served from the parent capture). effect="pure" is therefore a promise that re-running fn produces the same answer, and it is the one case a step body is checked for non-determinism - you asked for the check by declaring it.

retry (RetryPolicy) retries a failing attempt with backoff; timeout (seconds) bounds each attempt. Retries run live inside this call - only the final result/error is journaled, so replay is unchanged. A crash mid-retry re-runs the whole step on resume (standard at-least-once for the step’s side effects).

`once`

once(self, key: str, fn: Any, *args: Any, retry: Optional[RetryPolicy] = None, timeout: Optional[float] = None, **kwargs: Any) -> Any

Run fn at most once per idempotency key, durably - “don’t double-charge on retry”.

Unlike ctx.step (memoized per-run by journal position), once is keyed by key across runs: a retry, a resubmit of the whole workflow, or a second event for the same key returns the stored result without re-running fn. Backed by the KV store; journaled as a step so within-run replay is exact.

Guarantee: exactly-once within replay; idempotent across sequential retries/resubmits. Concurrent runs racing on a fresh key may both run fn (KV put is last-writer-wins, not a claim) - pass key into fn and hand it to your provider for provider-side idempotency if you need concurrent safety.

`gather`

gather(self, *legs: Awaitable[Any], return_exceptions: bool = False) -> list[Any]

Run legs concurrently and return their results in argument order.

Each leg is a coroutine that may call ctx.step/sleep/signal/etc. - Temporal’s asyncio.gather(execute_activity(a), execute_activity(b)) idiom. Durable and deterministic: each leg runs in an isolated replay scope, so concurrent steps (even with the same step_id) key distinctly and replay correctly. gather itself writes no event - the legs’ steps are the durable effects, so crash recovery just re-runs the body (completed steps memoize, incomplete re-run).

Pass legs as coroutines in a deterministic order::

results = await ctx.gather(fetch(ctx, a), fetch(ctx, b))

With return_exceptions=False (default) the first leg to raise cancels the siblings and the exception propagates; with return_exceptions=True the result list mixes values and exceptions (no cancellation), like asyncio.gather.

Legs must run concurrent work (ctx.step / run_child / deterministic), NOT suspending primitives (wait_for_signal / sleep / interrupt). The run has a single suspension slot, so two legs suspending at once cannot be represented; do the suspends sequentially around the gather instead.

`run_child`

run_child(self, name: str, params: Optional[dict[str, Any]] = None, poll_interval: float = 0.5, timeout: Optional[float] = None) -> Any

Run a child @workflow durably and return its result.

The child is a separate run (its own id + event log) - Temporal’s child workflow. Deterministic and crash-safe: the child run_id is derived from this call site, the submit is idempotent (a crash-and-resume re-attaches the SAME child instead of spawning a duplicate), and the result is journaled, so replay returns it without re-running the child. Composes with gather for parallel children::

a, b = await ctx.gather(ctx.run_child("wf_a", x), ctx.run_child("wf_b", y))

Raises ChildWorkflowError if the child ends in error or does not finish within timeout seconds (None waits indefinitely).

`send_signal`

send_signal(self, target_name: str, target_run_id: str, signal_name: str, payload: Optional[dict[str, Any]] = None) -> str

Durably send a signal to another run - Temporal’s signal-with-peer.

At-least-once delivery (as Temporal signals are): the send is journaled, so on the common path a crash-and-resume replay returns the original store-allocated signal_id without re-delivering. The one re-send window is a crash after the send fired but before its STEP_DONE persisted - replay then re-sends. Receivers should treat duplicates as idempotent. The target receives it via ctx.wait_for_signal(signal_name). Deterministic per call site, so it composes with gather for fan-out signalling::

await ctx.gather(*(ctx.send_signal("wf", rid, "go") for rid in workers))

`on_query`

on_query(self, name: str, handler: Callable[..., Any]) -> None

handler typically closes over the body’s local state::

@workflow
async def order(ctx, items):
    total = 0
    ctx.on_query("total", lambda: total)
    for it in items:
        total += await ctx.step(f"price:{it}", price, it)

A caller reads it via runner.query(name, run_id, "total"), which replays the run read-only (no new events) to re-register handlers and rebuild locals, then invokes the named handler. Registration itself never writes events and is never invoked on a normal run, so it has no determinism impact. Re-registering the same name replaces the prior handler (last write wins), so a handler registered inside a loop reflects the latest state.

`on_update`

on_update(self, name: str, handler: Callable[..., Any], validator: Optional[Callable[..., Any]] = None) -> None

Register an async update handler (+ optional sync validator) - Temporal’s update. Unlike on_query the handler MAY mutate the body’s locals and await ctx.step(...) / run_child(...); unlike send_signal it returns a value to the caller. A caller invokes it via runner.update(name, run_id, "apply", payload).

The handler runs at the next suspension-primitive frontier (see _process_pending_updates), in an isolated update:{seq} replay scope, and must NOT itself suspend (raises UpdateHandlerSuspended). The validator, if given, is sync + read-only and rejects the update before it enters history. Registration writes no event and is re-registered on replay (last write wins per name), so it has no determinism impact - same as on_query.

`continue_as_new`

continue_as_new(self, params: Optional[dict[str, Any]] = None, name: Optional[str] = None) -> Any

End this run and start a fresh run of the same workflow with params and an empty event log - Temporal’s continue_as_new, for bounding history growth in long-running loops (e.g. an agent conversation that never ends). Carry state forward by packing it into params; the successor starts clean.

Never returns: raises ContinueAsNewError, which the executor turns into a RUN_CONTINUED event for this run plus a successor run (idempotent submit). A parent awaiting this run via ctx.run_child follows the chain to the final result automatically.

`now`

now(self) -> float

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

`uuid`

uuid(self) -> str

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

`random`

random(self) -> float

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

`emit`

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

`patched`

patched(self, change_id: str) -> bool

Temporal-style branch gate for deploy-safe versioning.

Decides fresh-vs-replay off the run-level replay cursor (the same position-vs-completed-count signal the other primitives use), keyed to where this exact call lands in the journal - not a run-wide flag. Three cases:

A PATCH_MARKER for change_id is already recorded (present in the reconstructed journal, or written earlier this attempt) -> True.
No marker AND this call is reached during replay (the body is still inside the journaled prefix, state.is_replaying()): an old run that progressed past this point without ever calling patched. Return False (old branch) and write nothing - we are replaying.
No marker AND this call is reached live (journal exhausted, first real execution here): write a PATCH_MARKER and return True (new branch).

This makes a patch inserted before an already-completed step deterministic (reached before the cursor -> old branch) while a run that suspended before the patch and resumes into it (reached past the cursor -> new branch) takes the new branch - both stable across every replay.

`deprecate_patch`

deprecate_patch(self, change_id: str) -> None

Mark a patch as deprecated (no-op sentinel for future tooling).

Records the same PATCH_MARKER event so the patch still appears present in any replay that reaches this point. Call this before removing the old branch to signal that the patch guard is pending removal.

`emit_status`

emit_status(self, text: str) -> RunEvent

`sleep`

sleep(self, duration: timedelta | float) -> None

Timer re-queues after wake_time. Worker slot freed.

`sleep_until`

sleep_until(self, dt: datetime) -> None

Returns immediately if dt is already past.

`wait_for_signal`

wait_for_signal(self, name: str) -> Signal

`interrupt`

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

`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`

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`

A batch of records from a source.

Fields:

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

`CacheStore`

`cache_get`

cache_get(self, key: str) -> Optional[Any]

`cache_set`

cache_set(self, key: str, value: Any, ttl: float) -> None

`ChildWorkflowError`

A child run started with ctx.run_child ended in error or timed out.

`Collect`

Collect triggering events into one run (a batch trigger).

Events routed to the same resolved key accumulate; a run starts when the buffer reaches max_events OR window_seconds elapses since the first buffered event, whichever comes first. The run’s params are {"key": <key>, "events": [...], "count": <n>}. Named Collect (not Batch) to avoid clashing with the streaming Batch type.

Fields:

key: str | Callable[[dict], str]
max_events: int
window_seconds: float

`resolve`

resolve(self, params: dict) -> str

`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, 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`

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, idempotency_body_hash: Optional[str] = None, idempotency_ttl: int = 86400, version: str = '', version_behavior: str = 'pinned', concurrency_key: Optional[str] = None, concurrency_max: Optional[int] = None, rate_key: Optional[str] = None, rate_per_second: Optional[float] = None, rate_burst: Optional[int] = None) -> str

`claim_next_run`

claim_next_run(self, name: str, worker_id: str, worker_version: Optional[str] = None) -> Optional[str]

`claim_run`

claim_run(self, name: str, run_id: str, worker_id: str) -> None

`release_run`

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

`requeue_run`

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

`recover_pending_runs`

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

`get_pending_count`

get_pending_count(self, name: str) -> int

`heartbeat`

heartbeat(self, name: str, run_id: str, worker_id: str) -> None

`request_cancel`

request_cancel(self, name: str, run_id: str) -> str

`get_cancel_requested_at`

get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str]

`expire_run`

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

`append_event`

append_event(self, name: str, run_id: str, event: RunEvent) -> None

`append_events`

append_events(self, name: str, run_id: str, events: list[RunEvent]) -> None

`replay_run`

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

`load_run`

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

`get_last_event`

get_last_event(self, name: str, run_id: str) -> Optional[RunEvent]

`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`

set_run_owner(self, name: str, run_id: str, owner: str) -> None

`get_run_owner`

get_run_owner(self, name: str, run_id: str) -> Optional[str]

`ping`

ping(self) -> None

`list_runs`

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

`set_interrupted`

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

`resume_run`

resume_run(self, name: str, run_id: str, action: str, value: Any) -> None

`complete_run`

complete_run(self, name: str, run_id: str) -> None

`get_interrupt_response`

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

`get_idempotency_run`

get_idempotency_run(self, name: str, key: str) -> Optional[str]

`set_idempotency_run`

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

`cache_get`

cache_get(self, key: str) -> Optional[Any]

`cache_set`

cache_set(self, key: str, value: Any, ttl: float) -> None

`set_sleeping`

set_sleeping(self, name: str, run_id: str, wake_time: float) -> None

`get_sleeping_runs`

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

`wake_run`

wake_run(self, name: str, run_id: str) -> None

`kv_get`

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

`kv_put`

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

`kv_delete`

kv_delete(self, namespace: str, key: str) -> None

`kv_list`

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

`kv_search`

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

`send_signal`

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

`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`

set_signal_waiting(self, name: str, run_id: str, signal_name: str, call_index: int) -> None

`arm_signal_wait`

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

`deliver_update`

deliver_update(self, name: str, run_id: str, update_name: str, payload: dict[str, Any], update_id: str, sent_at: str) -> str

`drain_pending_updates`

drain_pending_updates(self, name: str, run_id: str) -> list[Any]

`write_update_result`

write_update_result(self, name: str, run_id: str, update_id: str, outcome: dict[str, Any]) -> None

`read_update_result`

read_update_result(self, name: str, run_id: str, update_id: str) -> Optional[dict[str, Any]]

`ContinuationChainError`

Raised by resolve_continuation (and any hand-rolled chain-follower that reuses its max_hops semantics, e.g. ctx.run_child) when a RUN_CONTINUED chain exceeds max_hops without resolving to a stable tail, or when a RUN_CONTINUED event carries a malformed continued_to (missing or not a string). A cycle (A -> B -> A), a runaway chain, or a corrupted continuation payload must fail loud - rather than spin forever walking store reads, or surface a raw KeyError/TypeError - so pass reason for the latter case.

Fields:

name = name
run_id = run_id
max_hops = max_hops
reason = reason

`Debounce`

Coalesce a burst of events into one run after they go quiet.

Events routed to the same resolved key accumulate; a run starts once window_seconds passes with no new event for that key (the timer resets on every event). Run params match :class:Collect.

Fields:

key: str | Callable[[dict], str]
window_seconds: float

`resolve`

resolve(self, params: dict) -> str

`DurableInterrupt`

A run driven by run_and_await suspended on ctx.interrupt and no on_interrupt handler was supplied. Carries the interrupt payload so the caller (e.g. an as_langgraph_node adapter) can surface it to its host.

Fields:

code = ErrorCode.SUSPEND
name = name
run_id = run_id
event_id = event_id
payload = payload

`DurableInvocationError`

A run driven by run_and_await ended in error.

Fields:

name = name
run_id = run_id
detail = detail

`DurableInvocationTimeout`

A run driven by run_and_await did not reach a terminal state in time.

Fields:

code = ErrorCode.RUN_TIMEOUT

`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_CONTINUED = 'run:continued'
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'
RUN_UPDATE_RECEIVED = 'run:update_received'
RUN_UPDATE_COMPLETED = 'run:update_completed'
RUN_UPDATE_FAILED = 'run:update_failed'
STEP_PENDING = 'step:pending'
STEP_DONE = 'step:done'
STEP_ERROR = 'step:error'
HEARTBEAT = 'heartbeat'
STREAM_OVERFLOW = 'stream:overflow'
PATCH_MARKER = 'patch_marker'

`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`

append_event(self, name: str, run_id: str, event: RunEvent) -> None

`append_events`

append_events(self, name: str, run_id: str, events: list[RunEvent]) -> None

Append events in order. Default loops over append_event; backends SHOULD override with a single-round-trip batch write. Callers (the run engine’s write-behind buffer) rely only on ordering, not atomicity: a partial batch on crash is indistinguishable from a crash between single appends.

`load_run`

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

`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`

get_last_event(self, name: str, run_id: str) -> Optional[RunEvent]

`read_entries_from`

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

`Fatal`

A non-retryable, unrecoverable failure (missing config, invariant broken). RetryPolicy never retries it; the run aborts.

`IdempotencyStore`

`get_idempotency_run`

get_idempotency_run(self, name: str, key: str) -> Optional[str]

`set_idempotency_run`

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

`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`

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

`seek`

seek(self, source_id: str, partition: int, offset: int) -> None

`get_committed`

get_committed(self, source_id: str, partition: int) -> Optional[int]

`partitions`

partitions(self) -> list[tuple[str, int]]

`inject_committed`

inject_committed(self, source_id: str, partition: int, offset: int) -> None

Test helper: set committed offset directly.

`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`

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

`trigger`

trigger(self, name: str, event: dict[str, Any]) -> Optional[str]

Route an event to name’s batch/debounce buffer. Returns the run id if this event completed a batch (fired immediately), else None (a time-based fire happens later via fire_due_triggers).

`fire_due_triggers`

fire_due_triggers(self, name: str, now: Optional[float] = None) -> list[str]

Submit a run for each of name’s buckets whose window has elapsed.

`run_next`

run_next(self, name: str) -> Optional[list[RunEvent]]

`run`

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

`update`

update(self, name: str, run_id: str, update_name: str, payload: Optional[dict[str, Any]] = None, update_id: Optional[str] = None, timeout: float = 30.0) -> Any

Deliver an update and return its handler’s value - Temporal’s update.

Delivers the update to the store, drives an attempt (which processes it at the next suspension-primitive frontier or completion), then reads the result slot. Raises UpdateRejected if a validator rejected it, UpdateError for an unknown handler / handler failure / no-result. Idempotent by update_id (a re-delivery returns the settled outcome without re-running the handler). timeout is accepted per the networked-op rule; the in-process drive is synchronous so it settles within the attempt, and the HTTP path enforces it on the await transport.

`query`

query(self, name: str, run_id: str, query_name: str, *args: Any, **kwargs: Any) -> Any

Read in-flight run state via a ctx.on_query handler (Temporal query).

Replays the run read-only and invokes the named handler; writes no events. Raises QueryError for an unknown query or a run that has not started.

`aclose`

aclose(self) -> None

`InterruptAction`

Fields:

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

`InterruptResponse`

Fields:

action: InterruptAction | str
value: Any = None

`KVStore`

`kv_get`

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

`kv_put`

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

`kv_delete`

kv_delete(self, namespace: str, key: str) -> None

`kv_list`

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

`kv_search`

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

`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`

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, idempotency_body_hash: Optional[str] = None, idempotency_ttl: int = 86400, version: str = '', version_behavior: str = 'pinned', concurrency_key: Optional[str] = None, concurrency_max: Optional[int] = None, rate_key: Optional[str] = None, rate_per_second: Optional[float] = None, rate_burst: Optional[int] = None) -> str

Submit a run for execution.

idempotency_body_hash/idempotency_ttl refine idempotency_key: when set, a resubmission under the same key is only treated as a cache hit if the body hash also matches, and the idempotency record expires after idempotency_ttl seconds (default 86400 = 1 day).

`claim_next_run`

claim_next_run(self, name: str, worker_id: str, worker_version: Optional[str] = None) -> Optional[str]

`claim_run`

claim_run(self, name: str, run_id: str, worker_id: str) -> None

Assert ownership of a specific run (force-set lock + deadline).

Distinct from claim_next_run (consume the queue): this claims the named run_id directly. Not a conditional acquire - the caller has already established it should own the run.

`release_run`

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

`requeue_run`

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

`recover_pending_runs`

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

`get_pending_count`

get_pending_count(self, name: str) -> int

`heartbeat`

heartbeat(self, name: str, run_id: str, worker_id: str) -> None

`request_cancel`

request_cancel(self, name: str, run_id: str) -> str

`get_cancel_requested_at`

get_cancel_requested_at(self, name: str, run_id: str) -> Optional[str]

`expire_run`

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

`set_run_owner`

set_run_owner(self, name: str, run_id: str, owner: str) -> None

`get_run_owner`

get_run_owner(self, name: str, run_id: str) -> Optional[str]

`get_run_version`

get_run_version(self, name: str, run_id: str) -> Optional[tuple[str, str]]

Return the (version, version_behavior) stamped at submit/adoption, or None when the store has no record for the run.

`adopt_run`

adopt_run(self, name: str, run_id: str, new_version: str) -> None

Re-stamp an in-flight run onto new_version (compatible upgrade).

`submitted_at`

submitted_at(self, name: str, run_id: str) -> Optional[float]

Epoch seconds when the run was submitted, or None if unknown.

`get_pause_requested_at`

get_pause_requested_at(self, name: str, run_id: str) -> Optional[str]

ISO timestamp of a pending pause request, or None. Pairs with get_cancel_requested_at.

`apply_retention`

apply_retention(self, name: str, run_id: str) -> None

Apply the store’s completion retention policy (TTL / eviction) to a finished run. Idempotent. Pairs with append_event_sequenced(..., retain=False), which appends a terminal event but defers retention so a coordinator (ArchivingStore) can archive first. The default raises so a backend asked for it without providing it fails loud, not silent (Rule 3).

`ping`

ping(self) -> None

`list_runs`

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

`set_interrupted`

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

`resume_run`

resume_run(self, name: str, run_id: str, action: str, value: Any) -> None

`complete_run`

complete_run(self, name: str, run_id: str) -> None

`get_interrupt_response`

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

`Limit`

A per-key concurrency limit.

key selects the fairness bucket: a param name ("tenant_id" -> params["tenant_id"]) or a callable params -> str for composite keys. max is the maximum number of runs sharing that key allowed to execute concurrently (“N actively executing” - a suspended run frees its slot).

Fields:

key: str | Callable[[dict], str]
max: int

`resolve`

resolve(self, params: dict) -> str

Resolve the concrete key string for a run’s params.

`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

`list_names`

list_names(self) -> list[str]

`Poison`

A non-retryable failure caused by bad input/data. RetryPolicy never retries it - retrying cannot help - so it fails fast (and routes to the DLQ/quarantine path) instead of burning the retry budget.

`PostgresArchive`

Postgres-backed durable archive for completed runs.

Stores only the completed event stream (history). In-flight runs, lifecycle, idempotency, signals, timers, and KV are the live tier’s concern - this is an archive, not a second live copy.

Schema is owned by Alembic; apply it with pyrula.workflows.stores.postgres.migrate(dsn) before use.

Fields:

dsn = dsn
replay_page_size = replay_page_size

`archive_run`

archive_run(self, name: str, run_id: str, events: list[tuple[str, RunEvent]]) -> None

Idempotently persist the completed run’s event stream.

events carries (stream_offset, RunEvent) tuples so the canonical run_seq/offset survive. The coordinator reads them from the live tier and hands them here - the archive never touches the live store.

`load_archived_run`

load_archived_run(self, name: str, run_id: str) -> list[RunEvent]

`load_archived_run_owner`

load_archived_run_owner(self, name: str, run_id: str) -> Optional[str]

`replay_archived_run`

replay_archived_run(self, name: str, run_id: str, cursor: Optional[Any], count: Optional[int]) -> tuple[list[RunEvent], Optional[Any]]

`read_archived_entries_from`

read_archived_entries_from(self, name: str, run_id: str, cursor: Optional[str], count: int) -> tuple[list[tuple[str, RunEvent]], Optional[str]]

`list_archived_names`

list_archived_names(self) -> list[str]

`list_archived_runs`

list_archived_runs(self, name: str) -> list[str]

`QueryError`

A synchronous run query failed - unknown query name, the run has not started, or its journal could not be replayed to serve the query.

Fields:

code = ErrorCode.INVALID_REQUEST

`Rate`

A per-key token-bucket rate limit.

key selects the bucket (same forms as :class:Limit). per_second is the sustained refill rate; burst is the bucket capacity (max tokens that can accumulate) and defaults to ceil(per_second) (min 1) - a one-second burst. Each claim of a run sharing the key consumes one token; a claim is denied (the run stays pending) when the bucket is empty.

Fields:

key: str | Callable[[dict], str]
per_second: float
burst: Optional[int] = None
capacity: int

`resolve`

resolve(self, params: dict) -> str

Resolve the concrete key string for a run’s params.

`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)
child_results: dict[str, Any] = field(default_factory=dict)
signal_send_results: dict[str, Any] = 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)
stream_committed_batch_counts: dict[tuple[str, int], int] = field(default_factory=dict)
patch_markers: set[str] = field(default_factory=set)
updates_by_frontier: dict[int, list[dict[str, Any]]] = field(default_factory=dict)
journaled_effect_count: int = 0

`RetryPolicy`

How ctx.step retries a failing attempt.

max_attempts counts the initial try (so 1 = no retry). Backoff grows geometrically from initial_backoff by backoff_multiplier, capped at max_backoff. If retryable is set, only those exception types are retried; anything else fails immediately. None (default) retries every Exception. Transient/Poison/Fatal (pyrula.contracts.errors) are honored regardless of retryable: Transient always retries (up to max_attempts), Poison/Fatal never do.

Fields:

max_attempts: int = 3
initial_backoff: float = 0.1
max_backoff: float = 30.0
backoff_multiplier: float = 2.0
retryable: Optional[tuple[type[BaseException], ...]] = None

`should_retry`

should_retry(self, attempt: int, exc: BaseException) -> bool

True if a failed attempt (1-based) should be retried for exc.

Poison/Fatal are never retried (retrying cannot help); Transient is always retryable up to max_attempts. Any other exception keeps the prior behavior: the retryable tuple gates it, or (None) every Exception retries.

`backoff_seconds`

backoff_seconds(self, attempt: int) -> float

Delay before the attempt after attempt (1-based).

`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`

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

`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`

run_attempt(self, name: str, run_id: str) -> list[RunEvent]

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

`RunRef`

Fields:

name: str
run_id: str

`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`

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

Called at event-write-buffer flush time, per event, in append order.

Thread affinity is NOT guaranteed: EventWriteBuffer.flush is “safe from any thread” and runs on either the event loop thread or an asyncio.to_thread worker, and it fires while the buffer’s per-run RLock is held. Implementations MUST be thread-safe and non-blocking, and MUST NOT call back into the run’s store wrapper - a re-entrant flush from inside on_event is a no-op (the RLock just makes it non-deadlocking), but preserving event ordering across such a re-entrant call is on the implementer, not the buffer.

`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`

wants_live_sequencing(self) -> bool

`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
event_buffer: Optional[EventWriteBuffer] = None
event_cap_check: Optional[Callable[[RunEvent], None]] = None
event_count_cap: Optional[int] = 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)
clock: Callable[[], float] = time.time
step_done_worker: Optional[str] = None
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)
gather_call_counts: dict[str, int] = field(default_factory=dict)
child_call_counts: dict[str, int] = field(default_factory=dict)
signal_send_call_counts: dict[str, int] = field(default_factory=dict)
query_handlers: dict[str, Callable[..., Any]] = field(default_factory=dict)
update_handlers: dict[str, tuple[Callable[..., Any], Optional[Callable[..., Any]]]] = field(default_factory=dict)
verify_mode: bool = False
verify_replay_log: Optional[list[tuple[str, int]]] = None
patch_markers: set[str] = field(default_factory=set)
replay_position: int = 0
update_frontier_index: int = 0

`is_replaying`

is_replaying(self) -> bool

True while the body is still re-consuming the journaled prefix.

Compares the run-level replay cursor (effects re-consumed this attempt) against the journaled effect count reconstructed from the run’s events - mirroring the per-effect position < completed_count test the other primitives use. False on a fresh run (no replay state) and once the journal is exhausted (the body is executing live).

`advance_replay_cursor`

advance_replay_cursor(self) -> None

Record that one journaled effect was served from the journal.

`scoped`

scoped(self, base: str) -> str

Build a scope-keyed replay base from the current task’s scope segments (a ContextVar). Empty scope returns base unchanged.

`write`

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

`write_async`

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

`poll_cancellation`

poll_cancellation(self) -> Optional[str]

`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`

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`

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

`Signal`

Fields:

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

`SignalStore`

`send_signal`

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

`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`

set_signal_waiting(self, name: str, run_id: str, signal_name: str, call_index: int) -> None

`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`

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`

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

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

`seek`

seek(self, source_id: str, partition: int, offset: int) -> None

Seek to a specific offset (used on resume).

`get_committed`

get_committed(self, source_id: str, partition: int) -> Optional[int]

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

`partitions`

partitions(self) -> list[tuple[str, int]]

Return all (source_id, partition) pairs.

`StepKind`

Fields:

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

`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`

Fields:

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

`TimerStore`

`set_sleeping`

set_sleeping(self, name: str, run_id: str, wake_time: float) -> None

`get_sleeping_runs`

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

`wake_run`

wake_run(self, name: str, run_id: str) -> None

`Transient`

A retryable failure: a passing condition (network blip, lock contention). RetryPolicy retries it up to max_attempts even without an explicit retryable.

`UnboundContextError`

`ValkeyStore`

Fields:

SLEEPING_INDEX_KEY: str = 'sleeping:index'
SCHEDULES_KEY: str = 'schedule:index'
SCHEDULE_FIRE_TTL: int = 300
url = url
completed_ttl = completed_ttl
maxlen = maxlen
replay_page_size = replay_page_size
heartbeat_interval = heartbeat_interval
lock_ttl = lock_ttl
max_replay_entries = 2 * maxlen if max_replay_entries is None else max_replay_entries
retry_attempts = retry_attempts
retry_base = retry_base
max_retries = max_retries
socket_timeout = socket_timeout
socket_connect_timeout = socket_connect_timeout

`from_url`

from_url(cls, url: str, **kwargs: Any) -> Self

`load_function_library`

load_function_library(self, source: str) -> None

`list_names`

list_names(self) -> list[str]

`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
projection_interval: float = 60.0
stranded_policy: StrandedPolicy = StrandedPolicy.WAIT
stranded_timeout: float = 0.0

`WorkflowContext`

Alias of 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 Whether cancellation has been requested for this run.

continue_suggested: bool True once this run’s event log has consumed >= 80% of its max_event_count_per_run cap (Temporal’s suggest-continue idiom): call ctx.continue_as_new at a convenient boundary when this turns True. Always False when the run is uncapped. Pure property, no I/O.

`checkpoint`

checkpoint(self) -> None

`step`

step(self, step_id: str, fn: Any, *args: Any, step_kind: str = StepKind.NAMED, effect: str = 'write', retry: Optional[RetryPolicy] = None, timeout: Optional[float] = None, **kwargs: Any) -> Any

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

inside a step, anything goes; outside one, the workflow body must be
deterministic - use ctx.now / ctx.uuid / ctx.random there.

Prefer ctx.sleep over time.sleep in a step: only ctx.sleep survives a crash, and a blocking sleep holds a worker thread for its duration.

step_id must be stable across retries. 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.

(1, 2)        -> [1, 2]            tuples become lists
{1: "a"}      -> {"1": "a"}        non-string dict keys become strings

`once`

once(self, key: str, fn: Any, *args: Any, retry: Optional[RetryPolicy] = None, timeout: Optional[float] = None, **kwargs: Any) -> Any

Run fn at most once per idempotency key, durably - “don’t double-charge on retry”.

`gather`

gather(self, *legs: Awaitable[Any], return_exceptions: bool = False) -> list[Any]

Run legs concurrently and return their results in argument order.

Pass legs as coroutines in a deterministic order::

results = await ctx.gather(fetch(ctx, a), fetch(ctx, b))

`run_child`

run_child(self, name: str, params: Optional[dict[str, Any]] = None, poll_interval: float = 0.5, timeout: Optional[float] = None) -> Any

Run a child @workflow durably and return its result.

a, b = await ctx.gather(ctx.run_child("wf_a", x), ctx.run_child("wf_b", y))

Raises ChildWorkflowError if the child ends in error or does not finish within timeout seconds (None waits indefinitely).

`send_signal`

send_signal(self, target_name: str, target_run_id: str, signal_name: str, payload: Optional[dict[str, Any]] = None) -> str

Durably send a signal to another run - Temporal’s signal-with-peer.

await ctx.gather(*(ctx.send_signal("wf", rid, "go") for rid in workers))

`on_query`

on_query(self, name: str, handler: Callable[..., Any]) -> None

handler typically closes over the body’s local state::

@workflow
async def order(ctx, items):
    total = 0
    ctx.on_query("total", lambda: total)
    for it in items:
        total += await ctx.step(f"price:{it}", price, it)

`on_update`

on_update(self, name: str, handler: Callable[..., Any], validator: Optional[Callable[..., Any]] = None) -> None

`continue_as_new`

continue_as_new(self, params: Optional[dict[str, Any]] = None, name: Optional[str] = None) -> Any

`now`

now(self) -> float

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

`uuid`

uuid(self) -> str

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

`random`

random(self) -> float

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

`emit`

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

`patched`

patched(self, change_id: str) -> bool

Temporal-style branch gate for deploy-safe versioning.

A PATCH_MARKER for change_id is already recorded (present in the reconstructed journal, or written earlier this attempt) -> True.
No marker AND this call is reached during replay (the body is still inside the journaled prefix, state.is_replaying()): an old run that progressed past this point without ever calling patched. Return False (old branch) and write nothing - we are replaying.
No marker AND this call is reached live (journal exhausted, first real execution here): write a PATCH_MARKER and return True (new branch).

`deprecate_patch`

deprecate_patch(self, change_id: str) -> None

Mark a patch as deprecated (no-op sentinel for future tooling).

`emit_status`

emit_status(self, text: str) -> RunEvent

`sleep`

sleep(self, duration: timedelta | float) -> None

Timer re-queues after wake_time. Worker slot freed.

`sleep_until`

sleep_until(self, dt: datetime) -> None

Returns immediately if dt is already past.

`wait_for_signal`

wait_for_signal(self, name: str) -> Signal

`interrupt`

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

`stream`

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

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

`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.

`WorkflowProfile`

Run-kind profile for workflows (no LLM).

`reconstruct`

reconstruct(self, events: list[RunEvent]) -> ReplayState

Rebuild replay state from the event log. Agents override this to use agent-specific arms (llm/tool) and AgentReplayState.

`make_state`

make_state(self, inputs: StateInputs) -> RunState

Assemble the run-kind’s state object from executor-built plumbing. Agents override this to build a TurnState (+ thread id, hide_thinking).

`continuation_metadata`

continuation_metadata(self, state: RunState) -> dict[str, Any]

Extra metadata to stamp on a continue_as_new successor’s RUN_INIT, merged after _continued_from/_chain_root. Workflows have no vocabulary to add here; agents override this to carry thread linkage forward without the executor knowing agent concepts.

`build_context`

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

`run_body`

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

`finalize`

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

`WorkflowSpec`

Fields:

name: str
timeout: float = 300.0
max_event_count_per_run: Optional[int] = None
max_event_payload_bytes: Optional[int] = None
max_attempt_wall_seconds: Optional[float] = None
version: str = ''
version_behavior: VersionBehavior = 'pinned'
concurrency: Optional[Limit] = None
rate_limit: Optional[Rate] = None
trigger: Optional[Collect | Debounce] = None

`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`

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`

run(self) -> None

Run all supervisor loops until stop() is called.

Constants

EVENT_SCHEMA_VERSION = '1'

Functions

`chain_root`

chain_root(metadata: dict[str, Any], run_id: str) -> str

The logical-turn id: metadata[‘_chain_root’] if present else run_id.

`current_run_state`

current_run_state() -> Optional[RunState]

The RunState of the currently executing run body, or None if not inside a run.

`data_step`

data_step(ctx: BaseContext, step_id: str, op: Any, *args: Any, **kwargs: Any) -> Any

Durable data operation with codec-driven replay.

`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`

detect_version_skew(state: RunState) -> Optional[dict[str, Any]]

`encode_result`

encode_result(obj: Any) -> dict[str, Any]

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

`make_archiving_postgres`

make_archiving_postgres(dsn: str, valkey_url: str, pool: Optional[Any] = None, **valkey_opts: Any) -> ArchivingStore

Compose the common live-Valkey + Postgres-archive pairing.

Convenience for ArchivingStore(live=ValkeyStore(...), archive=PostgresArchive(...)).

`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`

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

`resolve_continuation`

resolve_continuation(store: Any, name: str, run_id: str, max_hops: int = 100) -> tuple[str, str, Optional[RunEvent]]

Follow RUN_CONTINUED links from (name, run_id) to the chain tail.

Returns (tail_name, tail_run_id, tail_last_event). tail_name is the tail run’s workflow name (a hop MAY change the name, so callers that operate on the tail - error surfacing, interrupt resume, control ops - need it, not just the tail run id). tail_last_event is the tail’s current last event (None if the tail has no events yet - successor submitted but not started). Raises ContinuationChainError after max_hops (a cycle or runaway chain must fail loud, not spin), and also if a RUN_CONTINUED event’s continued_to is missing or not a string - a malformed chain link must fail loud too, not surface a raw KeyError/TypeError to the caller.

Two calling shapes, pick the one that matches the caller:

Point read (one-shot): a caller that just wants “the current terminal state of this run right now” - error surfacing, a status query, a control op - calls this once against the run’s own (name, run_id) and uses the result. There is no “next call” to worry about.
Poll loop (repeated calls over time): a caller that polls until a run reaches a terminal state - ctx.run_child, run_and_await, gather’s awaited legs - MUST cursor-advance: resolve once, then on every later poll call this again starting from the LAST RESOLVED (tail_name, tail_run_id), not from the run’s original (name, run_id). Caching/swapping in a resolved tail across polls is the CORRECT pattern here, not a shortcut to avoid - see gather’s swap-in-place of its awaited (name, run_id) pair for the reference implementation. Re-walking from the ORIGINAL starting point on every poll instead re-reads every hop between origin and tail on every single call; interior hops can legitimately disappear later (archive retention on ArchivingStore, or anything else that ages out old rows), and once one does, a re-walk-from-origin poller breaks even though the run is still live at the tail. A cursor-advance poller never asks about a hop again once it has moved past it, so it keeps working regardless of what happens upstream of the cursor.

Cost: O(hops-since-the-caller’s-starting-point) get_last_event store reads per call - O(hops) for a one-shot point read from the origin, O(1) amortized per poll for a cursor-advancing poll loop (the common case: no new hop happened since the last poll, so the call is one read that returns immediately).

`run_and_await`

run_and_await(store: Store, name: str, params: Optional[dict[str, Any]] = None, idempotency_key: str, owner: Optional[str] = None, on_interrupt: Optional[InterruptHandler] = None, poll_interval: float = 0.5, timeout: Optional[float] = None) -> Any

Submit name (idempotently, run_id = idempotency_key) and await its result.

Returns the run’s result. Raises DurableInvocationError if the run errored, DurableInvocationTimeout on deadline, and DurableInterrupt if the run suspends on ctx.interrupt with no on_interrupt handler. When a handler is given, it is called with the interrupt payload; its return value resumes the run (action "respond") and awaiting continues to the final result. A handler that raises propagates (the adapter’s escape hatch - e.g. bridging to a host’s own interrupt/resume).

Idempotent: re-invoking with the same idempotency_key re-attaches the same run (the store dedups the submit) rather than spawning a duplicate. If the run calls continue_as_new the chain is followed to the successor’s final result.

Must not run inside the worker that executes name - this awaits an external executor via the store.

`workflow`

workflow(fn: Optional[Callable[..., Any]] = None, name: Optional[str] = None, timeout: float = 300.0, max_event_count_per_run: Optional[int] = None, max_event_payload_bytes: Optional[int] = None, max_attempt_wall_seconds: Optional[float] = None, profile: Optional[WorkflowProfile] = None, version: Optional[str] = None, version_behavior: VersionBehavior = 'pinned', concurrency: Optional[Limit] = None, rate_limit: Optional[Rate] = None, trigger: Optional[Collect | Debounce] = None) -> Any

Decorator to register a function as a durable workflow.

Usage::

@workflow
async def my_workflow(ctx):
    ...

@workflow(name="custom", timeout=60)
async def my_workflow(ctx):
    ...

max_event_count_per_run / max_event_payload_bytes cap the run’s persisted event stream.

max_attempt_wall_seconds caps ONE execution attempt, not the run. It is min’d with timeout, so it can only lower the effective per-attempt deadline - the point being that timeout is the workflow author’s declaration while this is an operator-imposed ceiling layered over it (the agent runtime feeds its max_turn_wall_seconds limit in here).

A retried run can therefore exceed it in total, once per attempt. There is deliberately no whole-run wall budget: enforcing one would mean carrying accumulated elapsed time across attempts in the event log, and nothing needs it yet. This parameter was called max_run_wall_seconds until 2026-07-30, which read as exactly the cross-attempt budget it is not.