Pre-release: the packages are not on PyPI yet, so pip install does not resolve. Build from source to try it today.

Workflows

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

Types

`ArchivingStore`

Composes a live Store with a durable Archive.

Live-first reads with archive fallback. On terminal append, defers live retention, archives the completed stream, then applies retention - crash-safe via a KV pending set kept in the live tier.

The live store must support sequenced append (append_event_sequenced / append_events_sequenced), not just the plain event log; one offering only EventLogStore raises AttributeError on its first terminal.

Fields:

rebalance_shards: Callable[[str, str], list[int]]
release_shards: Callable[[str, str], None]

`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, ttl: Optional[int] = None) -> None

`peek_signal_or_arm`

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

`pop_signal_match`

pop_signal_match(self, name: str, run_id: str, signal_name: str, signal_id: str) -> bool

`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, prefer: Sequence[int] = ()) -> 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

`peek_pending_updates`

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

`ack_pending_update`

ack_pending_update(self, name: str, run_id: str, update_id: str) -> bool

`expire_run`

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

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

get_idempotency_body_hash(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, owner: Optional[str] = None) -> tuple[list[dict[str, Any]], Optional[str]]

Runs from both tiers, newest first. A finished run leaves the live tier shortly after it is archived, so listing only the live tier would drop it from view while its history is still here.

Both tiers page by the same keyset, so one page takes up to limit from each and keeps the newest limit of the union. A run in both - inside the window before its live copy expires - is shown once, with the live tier’s status, which is the fresher.

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

resume_paused_run(self, name: str, run_id: str, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

`resume_run`

resume_run(self, name: str, run_id: str, action: str, value: Any, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

`send_signal`

send_signal(self, name: str, run_id: str, signal_name: str, payload: dict[str, Any], sent_at: str, idempotency_key: Optional[str] = None) -> 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_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, provenance: Optional[dict[str, Any]] = None) -> str

`wake_run`

wake_run(self, name: str, run_id: str, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

`wake_due`

wake_due(self, name: str, run_id: str, due_by: float) -> WaitOutcome

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

`get_store_binding`

get_store_binding(self) -> Optional[StoreBinding]

`bind_store`

bind_store(self, binding: StoreBinding) -> StoreBinding

`clear_store_binding`

clear_store_binding(self, expected: StoreBinding) -> bool

`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 the stored result on replay without re-executing.

Non-determinism (clocks, uuids, I/O) belongs inside a step; outside one the workflow body must be deterministic (use ctx.now / ctx.uuid / ctx.random). Prefer ctx.sleep over time.sleep - only ctx.sleep survives a crash. step_id must be stable across retries; sync callables run in a thread, async callables are awaited.

The result is stored as JSON: tuples become lists, non-string dict keys become strings, models/dataclasses become dicts, datetime/Decimal/UUID become strings, set a list. A value with no JSON projection (bytes, timedelta, NaN) fails the step rather than recording something that can’t round-trip.

effect is “read” | “write” (default) | “pure”: the cloud uses it for a counterfactual branch’s per-step replay default (read/pure re-run live, write serves from the parent capture), and “pure” also enables a determinism check on fn. retry/timeout bound a live attempt - only the final result/error is journaled, so a crash mid-retry re-runs the whole step on resume.

`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 across sequential retries and resubmits - “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, or a repeat 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.

Not protected: a crash between fn returning and the result being stored re-runs fn on recovery, same as a bare write step; two runs racing on a fresh key may also both run fn (KV put is last-writer-wins, not a claim). Pass key into fn as the provider’s own idempotency key for an exactly-once guarantee that survives a crash.

`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 re-runs the body; completed steps return recorded results and incomplete steps execute again.

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.

Effectively once: the send is journaled, so on the common path a crash-and-resume replay returns the original result without re-delivering; and a crash after the send fired but before its STEP_DONE persisted re-sends under the same idempotency key, which the store deduplicates. A signal to a run that has already ended is not buffered and returns "rejected". 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 with await run.query("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 with await run.update("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. dt must be timezone-aware: a naive one means local time to some readers and UTC to others, so it is refused rather than guessed.

`wait_for_signal`

wait_for_signal(self, name: str) -> Signal

`interrupt`

interrupt(self, event_id: str, payload: dict[str, Any], expects: Optional[InterruptExpects] = None, options: Optional[list[str | dict[str, Any]]] = None) -> InterruptResponse

Suspend the run awaiting a human response.

payload is the reviewer-facing context. expects declares the response affordance a UI should draw - approval (yes/no), choice (pick one of options), text, or value - so it need not infer it from the payload. Both ride as siblings of payload on the event and are omitted when unset, which keeps a reviewer UI’s decision (approve/reject vs a free-form box) out of the domain data. options entries are label strings or {label, value}.

`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

Record the next offset so replay skips this batch.

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.

Ending a wait needs both roles on one backend and raises NotImplementedError otherwise.

`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, provenance: Optional[dict[str, Any]] = None) -> str

`claim_next_run`

claim_next_run(self, name: str, worker_id: str, worker_version: Optional[str] = None, prefer: Sequence[int] = ()) -> 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]

`request_pause`

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

`get_pause_requested_at`

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

`clear_pause`

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

`resume_paused_run`

resume_paused_run(self, name: str, run_id: str, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

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

`get_store_binding`

get_store_binding(self) -> Optional[StoreBinding]

`bind_store`

bind_store(self, binding: StoreBinding) -> StoreBinding

`clear_store_binding`

clear_store_binding(self, expected: StoreBinding) -> bool

`ping`

ping(self) -> None

`list_runs`

list_runs(self, name: str, limit: int = 50, cursor: Optional[str] = None, owner: 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, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

`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

`get_idempotency_body_hash`

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

`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, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

`wake_due`

wake_due(self, name: str, run_id: str, due_by: float) -> WaitOutcome

`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, idempotency_key: Optional[str] = None) -> str

`peek_signal_or_arm`

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

`pop_signal_match`

pop_signal_match(self, name: str, run_id: str, signal_name: str, signal_id: str) -> bool

`deliver_update`

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

`peek_pending_updates`

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

`ack_pending_update`

ack_pending_update(self, name: str, run_id: str, update_id: str) -> bool

`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 chain-follower reusing its max_hops semantics) when a RUN_CONTINUED chain exceeds max_hops without resolving to a stable tail, or carries a malformed continued_to. Fails loud rather than spin forever or surface a raw KeyError/TypeError; pass reason for the malformed-event 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 can surface it to its host. wait_id names the specific interrupt instance; pass it along when resuming rather than trusting “the run’s current interrupt”.

Fields:

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

`DurableInvocationCancelled`

A run driven by run_and_await ended because a cancel was requested on the store, not because it errored. A distinct type from DurableInvocationError so a caller (as_langgraph_node, the MCP tools server) can tell “the run was cancelled” from “the run failed” instead of catching one generic exception for both.

Fields:

code = ErrorCode.CANCELLED

`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_UNPAUSED = 'run:unpaused'
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 MUST make this atomic or idempotent under retry: the run engine’s write-behind buffer re-queues a failed batch and retries it verbatim, so a backend that commits a prefix and then raises will see that prefix again. The default loop satisfies neither property and is only acceptable for backends whose single append cannot partially fail.

Composing wrappers that override append_events (ArchivingStore, the agents LifecycleAdapter) must preserve that same property rather than reintroducing a per-event loop that re-exposes the bug the leaf stores already closed.

`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

`get_idempotency_body_hash`

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

The request body hash stored with key by submit_run, for telling a retry from a different request reusing the key. A store that records none returns None, and no conflict can be reported.

`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. capture, left unset, has no ceiling of its own: a workflow’s own declaration still wins where there is one, and otherwise PYRULA_CLOUD_CAPTURE sets the level outright, falling back to lifecycle only when neither is set. An explicit capture= here is a worker-wide ceiling, the same as a declaration would be, that the env var can only lower.

Fields:

client: WorkflowClient A WorkflowClient whose runs this runner advances in-process: a handle’s result, update and signal drive the run here instead of waiting for a worker.
store_binding_mismatched: bool

`fire_due_triggers`

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

Submit a run for each of name’s trigger 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]]

`aclose`

aclose(self) -> None

`InterruptAction`

The human’s decision on an interrupt, keyed to what it expects: approval -> APPROVE/REJECT; choice/text/value -> RESPOND (value carries the answer); an editable proposed action -> APPROVE/REJECT/EDIT. One vocabulary shared with the cloud control plane, which rejects anything else.

Fields:

APPROVE = 'approve'
REJECT = 'reject'
RESPOND = 'respond'
EDIT = 'edit'

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

Minimal claim and lease coordination interface. 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, provenance: Optional[dict[str, Any]] = 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).

The optional tail is keyword-only. Seven of the thirteen share a type (Optional[str] or Optional[int]), so a positional forwarder that transposed two of them type-checked and passed every test while swapping, say, the idempotency key with its body hash. Keyword-only makes that a TypeError at the call instead.

provenance is written with the run, atomically with its creation - a RunProvenance.to_dict() recording why it exists (trigger, parent run, attempt), for a run spawned from another to carry that lineage forward.

`claim_next_run`

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

Claim the next eligible run. prefer names shards to try first - the caller’s leased shards - before any other; a store without shards ignores it. A preference, never a restriction: the claim still reaches every shard, so work queued where no live worker holds a lease is never stranded.

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

`get_store_binding`

get_store_binding(self) -> Optional[StoreBinding]

Return the binding, or None for an unbound store.

Raises NotImplementedError when the store cannot hold a binding.

`bind_store`

bind_store(self, binding: StoreBinding) -> StoreBinding

Atomically bind an unbound store and return the stored winner.

`clear_store_binding`

clear_store_binding(self, expected: StoreBinding) -> bool

Remove the binding if it equals expected; return whether it was deleted.

`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, ttl: Optional[int] = None) -> None

Apply the store’s completion retention policy (TTL / eviction) to a finished run. Idempotent. ttl overrides the store’s own retention - how the archiving store keeps an archived run’s live copy briefly. 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, owner: Optional[str] = None) -> tuple[list[dict[str, Any]], Optional[str]]

Runs of name, newest first. With owner, only that owner’s - the web API scopes every listing to the authenticated principal.

`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, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

Answer the interrupt named wait_id and re-queue the run, only if that interrupt is still open. idempotency_key names one request; a repeat with the same key after it resolved returns DUPLICATE.

`resume_paused_run`

resume_paused_run(self, name: str, run_id: str, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

Lift the pause named wait_id and re-queue the run, only if that pause is still open. idempotency_key names one request; a repeat with the same key after it resolved returns DUPLICATE.

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

`NestedDurableOperationError`

Durable orchestration attempted inside an atomic step or tool.

Fields:

operation = operation
operation_name = operation_name
boundary_kind = boundary_kind
boundary_name = boundary_name

`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. Tracked in pyrula_archive_alembic_version rather than Alembic’s default alembic_version, so a host application sharing this database and running its own Alembic-managed schema never collides with it.

The first successful database use checks that the connected schema’s stamped revision matches what this pyrula version expects, raising ArchiveSchemaMismatch (not StoreUnavailable) on a mismatch - a missing or stale schema is a deploy error, not a condition worth retrying. check_schema() runs the same check on demand.

Fields:

dsn = dsn
replay_page_size = replay_page_size

`close`

close(self) -> None

Close the connections this archive opened itself. A pool it was handed belongs to the caller.

`check_schema`

check_schema(self) -> None

Raise ArchiveSchemaMismatch if the connected database’s archive schema is missing or stamped with a different revision than this pyrula version expects; returns None on a match.

Forces a fresh check regardless of whether an earlier automatic check on this instance already passed - the automatic guard caches success to keep the hot path cheap, but a caller that asks for this explicitly (pyrula db check) wants the current answer.

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

`list_archived_run_summaries`

list_archived_run_summaries(self, name: str, limit: int, after: Optional[tuple[float, str]] = None, owner: Optional[str] = None) -> list[tuple[float, str, str]]

(started_at, run_id, status) for archived runs of name, newest first, strictly after after in that order - a keyset page, so the archiving store can page this in step with the live tier.

`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[tuple[str, int], 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)
journaled_signal_ids: set[str] = field(default_factory=set)
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_update_ids: set[str] = field(default_factory=set)
journaled_update_outcomes: dict[str, dict[str, Any]] = field(default_factory=dict)
journaled_effect_count: int = 0

`RetryPolicy`

Configure retries for a failing ctx.step 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 counterfactual 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.

`RunFinishedError`

The run has reached a terminal event, so it takes no more signals or updates.

Fields:

code = ErrorCode.INVALID_REQUEST

`RunHandle`

One run: what WorkflowClient.submit returns and WorkflowClient.handle gets.

Each method is the client’s method of the same name, bound to this run.

Fields:

name = name
run_id = run_id

`signal`

signal(self, signal: str, payload: Optional[dict[str, Any]] = None, signal_id: Optional[str] = None) -> str

`query`

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

`update`

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

`cancel`

cancel(self) -> None

`resume`

resume(self, action: str, value: Any = None, wait_id: Optional[str] = None) -> None

`status`

status(self) -> RunStatus

`result`

result(self, timeout: Optional[float] = None, on_interrupt: Any = None) -> Any

`events`

events(self) -> list[RunEvent]

`RunNotFoundError`

No run with this name and id exists in the store.

Fields:

code = ErrorCode.NOT_FOUND

`RunNotWaitingError`

A resume named an interrupt the run is no longer waiting on.

Fields:

code = ErrorCode.INVALID_REQUEST

`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 to avoid deadlock; the implementer remains responsible for event ordering across such a call.

`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)
interrupt_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
update_ids_in_flight: set[str] = field(default_factory=set)

`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, idempotency_key: Optional[str] = None) -> str

Buffer a signal for the run and re-queue it if it is waiting on this name. Returns "delivered", "buffered", or "rejected" when the run has already ended (the signal is not kept).

idempotency_key names one send: a repeat with the same key returns the first send’s result and buffers nothing, so a redelivered signal is consumed once. A send without a key is buffered every time.

`peek_signal_or_arm`

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

Atomically peek a buffered signal (without popping it) if queued, else arm the waiting marker. Returns the buffered signal dict (deliver inline; carries signal_id for pop_signal_match) or None (suspend). One op so a concurrent send_signal cannot race the buffer-check/arm and lose the signal. The pop is a separate, guarded step (pop_signal_match) so a caller can journal the delivery durably before consuming the entry - a crash between peek and pop leaves the signal buffered for redelivery rather than losing it.

`pop_signal_match`

pop_signal_match(self, name: str, run_id: str, signal_name: str, signal_id: str) -> bool

Pop the head of the buffer only if its signal_id matches. Returns True when consumed, False when the head has since moved.

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

`StepContext`

Context available inside a decorated step.

Fields:

run_id: str
run_name: str
step_id: str
call_index: int
attempt: int
metadata: Mapping[str, Any]
deps: DepsT | None
principal: Any
idempotency_key: str
cancelled: bool

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

`StepResultTypeError`

Step result does not match its declared return type.

Fields:

step_id = step_id
validation_error = validation_error

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

`StoreBinding`

The project and environment a store belongs to.

bound_by identifies the key or operator; bound_at is ISO-8601 UTC. Destination matching compares only project and environment.

Fields:

project_id: str
environment: str
bound_by: str
bound_at: str

`belongs_to`

belongs_to(self, project_id: str, environment: str) -> bool

`to_json`

to_json(self) -> str

Canonical, so a compare-and-delete can compare stored text.

`from_json`

from_json(cls, raw: str) -> Self

`now`

now(cls, project_id: str, environment: str, bound_by: str) -> Self

`StoreBindingMismatch`

This process’s cloud key belongs to another project or environment than the run store it is attached to, so it will not run that store’s work. pyrula store binding show names the store’s binding.

`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, wait_id: str, idempotency_key: Optional[str] = None) -> WaitOutcome

End the sleep named wait_id and re-queue the run, only if that sleep is still open. idempotency_key names one request; a repeat with the same key after it resolved returns DUPLICATE.

`wake_due`

wake_due(self, name: str, run_id: str, due_by: float) -> WaitOutcome

End the run’s open sleep if its wake time is at or before due_by: the timer’s path, which knows only the run. If another wake re-arms the run after the timer reads its index, the timer finds that the new sleep is not due and leaves it open.

`Transient`

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

`UnboundContextError`

`UpdateError`

An update could not be applied: unknown handler, or the handler raised.

`UpdateRejected`

A registered validator rejected the update before it entered history.

`UpdateTimeout`

No worker applied the update within the caller’s timeout. It stays queued and may still apply; retry with the same update_id to read its outcome.

`ValkeyStore`

Durable run/turn store backed by Valkey.

namespace (constructor keyword, also PYRULA_VALKEY_NAMESPACE) scopes every key this store touches with f"{namespace}:" (see KeySpace), so one Valkey server can host several independent stores. Every store has a namespace - unset means "default", not “no namespace”. Read it back via the namespace property.

Fields:

SCHEDULE_FIRE_TTL: int = 300
namespace: str This store’s key namespace. Every store has one - "default" unless namespace= or PYRULA_VALKEY_NAMESPACE says otherwise.
SLEEPING_INDEX_KEY: str
SCHEDULES_KEY: str
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
shard_count = shard_count
shard_lease_ttl = shard_lease_ttl
cluster = cluster

`from_url`

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

`load_function_library`

load_function_library(self, source: str) -> None

`prune_stale_function_libraries`

prune_stale_function_libraries(self) -> list[str]

Delete every pyrula library on the server except this build’s. An operator step for after a rolling deploy finishes, never automatic: a process still on the previous build is still calling the library it loaded, and deleting it would turn its next store call into StoreUnavailable. The library is loaded once per server, not once per namespace, so this affects every namespaced store the server hosts.

`list_names`

list_names(self) -> list[str]

`shard_of`

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

The shard a run lives in: its gate’s if it was submitted with one, otherwise the one its run id hashes to.

`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
shard_lease_interval: float = 10.0

`WorkflowClient`

Start and interact with workflow runs through a store.

The caller-side API: a worker executes runs, a client starts them and talks to them. Every method is async and keeps store I/O off the event loop. Signal, query, update, cancel, resume, status and result act on the live tail of a continue_as_new chain, so a caller can keep using the id it started with.

Fields:

store: Store

`submit`

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

Queue a new run of name and return a handle to it.

With idempotency_key, a repeat submission returns the first run’s id instead of starting another. owner tags the run with a tenant for fair claim scheduling. The workflow’s version and flow-control policy are stamped from its registered spec.

`handle`

handle(self, name: str, run_id: str) -> RunHandle

A handle to an existing run.

`trigger`

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

Send event to name’s Collect or Debounce buffer.

Returns a handle to the run this event started by completing a batch, else None; a time-based fire is submitted later by a worker’s trigger loop.

`signal`

signal(self, name: str, run_id: str, signal: str, payload: Optional[dict[str, Any]] = None, signal_id: Optional[str] = None) -> str

Send a signal; returns "delivered" or "buffered".

signal_id makes a retried send idempotent. Raises RunFinishedError if the run has ended.

`query`

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

Call a ctx.on_query handler against the run’s replayed state.

Read-only: replays the run and writes nothing. Works on finished runs too. Raises QueryError for an unknown query.

`update`

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

Deliver an update, wait for a worker to apply it, and return its value.

update_id makes a retry idempotent: it returns the settled outcome without running the handler again. Raises UpdateRejected if the validator refused it, UpdateError if the handler failed, UpdateTimeout if no worker applied it within timeout, and RunFinishedError if the run has ended.

`cancel`

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

Ask the run to stop; it ends with a cancelled run:error at its next step.

`resume`

resume(self, name: str, run_id: str, action: str, value: Any = None, wait_id: Optional[str] = None) -> None

Answer the run’s ctx.interrupt with action (approve, reject, respond or edit) and value.

wait_id names the interrupt being answered, so a stale answer can’t land on a later one; it defaults to the interrupt the run is waiting on now. Raises RunNotWaitingError if the run is not waiting on it.

`status`

status(self, name: str, run_id: str) -> RunStatus

The run’s current status.

`result`

result(self, name: str, run_id: str, timeout: Optional[float] = None, on_interrupt: Any = None) -> Any

Wait for the run to finish and return its result.

Raises DurableInvocationError if it failed, DurableInvocationCancelled if it was cancelled, DurableInvocationTimeout past timeout, and DurableInterrupt if it waits on ctx.interrupt with no on_interrupt handler to answer it.

`events`

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

The run’s own journal, oldest first (not followed across continue_as_new).

`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 the stored result on replay without re-executing.

`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 across sequential retries and resubmits - “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. dt must be timezone-aware: a naive one means local time to some readers and UTC to others, so it is refused rather than guessed.

`wait_for_signal`

wait_for_signal(self, name: str) -> Signal

`interrupt`

interrupt(self, event_id: str, payload: dict[str, Any], expects: Optional[InterruptExpects] = None, options: Optional[list[str | dict[str, Any]]] = None) -> InterruptResponse

Suspend the run awaiting a human response.

`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

Record the next offset so replay skips this batch.

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

Fields:

run_kind: str = RunKind.WORKFLOW The run-kind this profile builds state for. A specialization (e.g. the agent profile in pyrula-agents) overrides this; the generic cloud advertisement in _connect_reporter skips any registry entry whose profile.run_kind isn’t RunKind.WORKFLOW, since the specialization advertises those runners itself - otherwise a combined deployment would have the workflow connector clobber the specialization’s registration.

`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
capture: Optional[CaptureLevel] = 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.

Fields:

client: WorkflowClient A WorkflowClient over this worker’s store.

Its submit raises BackpressureRejected while this worker’s cloud egress is behind: accepting work the cloud can’t be told about is how a backlog becomes unbounded, and a refusal the caller can retry is the honest form.

`refuse_submission`

refuse_submission(self) -> bool

Whether new work should be turned away right now (see WorkerSupervisorBase.refuse_submission). Declared directly because the generated reference does not render inherited members.

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

`detect_version_skew`

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

Compare live replay-derived counts against the journaled expectation; return the first mismatch found (or None), as {“expected”: …, “observed”: …}.

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

Load a run’s full event log, capped at the store’s own replay limit.

`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_last_event is None if the tail has no events yet. Raises ContinuationChainError past max_hops (cycle/runaway chain) or on a RUN_CONTINUED event with a missing/non-string continued_to. A one-shot point read calls this once from (name, run_id); a poll loop (run_child, run_and_await, gather) must cursor-advance - resolve once, then poll again from the last resolved tail, not the original run, since interior hops can age out and a re-walk from origin would then break a still-live tail. Cost is O(hops) per call from the given starting point.

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

Raises DurableInvocationCancelled if the run was cancelled, plain DurableInvocationError for any other run error, DurableInvocationTimeout on deadline, and DurableInterrupt if the run suspends on ctx.interrupt with no on_interrupt handler. With a handler, it’s called with the interrupt payload and its return value resumes the run; a handler that raises propagates (the adapter’s escape hatch).

Idempotent: re-invoking with the same idempotency_key re-attaches the same run rather than spawning a duplicate; a continue_as_new 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.

`step`

step(fn: Callable[..., Any] | None = None, name: str | None = None, effect: Effect = 'write', retry: RetryPolicy | None = None, timeout: float | None = None) -> Any

Define an atomic durable workflow step.

`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, capture: Optional[CaptureLevel | str] = 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: timeout is the workflow author’s declaration, this is an operator-imposed ceiling over it (the agent runtime feeds its max_turn_wall_seconds limit in here).

capture is how much of this workflow’s event stream leaves the process for the cloud - the same four levels an @agent takes, resolved the same way: a declaration here is a ceiling, and PYRULA_CLOUD_CAPTURE can only lower it, never raise it past what’s declared. none ships structure only - what ran, how long, pass or fail, ids, usage - and withholds everything else. lifecycle, the default, adds the workflow’s own data: a step’s own input and result, run params, payloads, status and error text. io adds model and tool traffic: an LLM step’s message, a tool or sub-agent call’s input and result. Left unset, the runner’s process-wide setting applies.

none is not total silence: the run’s own start and terminal events are reported by the reporter’s lifecycle calls, not the captured-event path, so the cloud still learns that the run started and how it ended.

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.